Nacos 服务列表管理
Nacos 提供了开放 API 可通过 /nacos/v1/ns/instance/list 获取服务列表。如果我们采用 spring-cloud 方式去获取服务，最终会通过 Nacos Client loadbalancer 的方式进行客户端负载均衡。
Ribbon 源码解析
Ribbon 简介
Spring Cloud Ribbon 是 Netflix Ribbon 实现的一套客户端负载均衡工具 简单的说，Ribbon 是 Netflix 发布的开源项目，主要功能是提供客户端的复杂算法和服务调用。 Ribbon 客户端组件提供一系列完善的配置项如超时、重试等。简单的说，就是配置文件中列出 load Balancer （简称 LB）后面所有的机器，Ribbon 会自动的帮助你基于某种规则（如简单轮询，随机链接等）去链接这些机器。我们很容易使用 Ribbon 自定义的负载均衡算法。
Ribbon 使用
首先需要定义 RestTemplate 使用 Ribbon 策略；
代码语言：javascript复制@Configuration
public class RestTemplateConfig {
@LoadBalanced
@Bean
public RestTemplate restTemplate() {
return new RestTemplate();
}
}
本地使用 RestTemplate 调用远程接口；
代码语言：javascript复制@Autowired
private RestTemplate restTemplate;
@RequestMapping(value = "/echo/{id}", method = RequestMethod.GET)
public String echo(@PathVariable Long id) {
return restTemplate.getForObject("http://member-service/member/get/" id, String.class);
}
Ribbon 源码分析
RestTemplate 继承 InterceptingHttpAccessor 通过 interceptors 字段接受 HttpRequestInterceptor 请求拦截器。对于 Ribbion 初始化类是 RibbonAutoConfiguration 中的， 它在 spring-cloud-netflix-ribbon 中定义。但是它在初始化之前，又需要加载 RibbonAutoConfiguration 配置，它是在 spring-cloud-common 中。具体的代码如下:
代码语言：javascript复制@Configuration(proxyBeanMethods = false)
// 工程中一定存在 RestTemplate 类
@ConditionalOnClass(RestTemplate.class)
// 容器中一定存在 LoadBalancerClient 类 Bean 实例
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {
// 获取 Spring 容器中所有的 RestTemplate 实例
@LoadBalanced
@Autowired(required = false)
private List<RestTemplate> restTemplates = Collections.emptyList();
// 获取 Spring 容器中 LoadBalancerRequestTransformer 实例
@Autowired(required = false)
private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();
// 在 Bean 初始化完成后会调用 afterSingletonsInstantiated 方法
// 这里是一个 lambda 表达式方式的实现， 主要是为 restTemplate 实例设置 RestTemplateCustomizer
@Bean
public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(
final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
return () -> restTemplateCustomizers.ifAvailable(customizers -> {
for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
for (RestTemplateCustomizer customizer : customizers) {
customizer.customize(restTemplate);
}
}
});
}
// LoadBalancerRequestFactory 工厂类
// 主要是用来提供 LoadBalancerClient 实例和 LoadBalancerRequestTransformer
@Bean
@ConditionalOnMissingBean
public LoadBalancerRequestFactory loadBalancerRequestFactory(
LoadBalancerClient loadBalancerClient) {
return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers);
}
// LoadBalancerInterceptor 拦截器
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")
static class LoadBalancerInterceptorConfig {
// 创建默认的拦截器 LoadBalancerInterceptor 的实例
@Bean
public LoadBalancerInterceptor ribbonInterceptor(
LoadBalancerClient loadBalancerClient,
LoadBalancerRequestFactory requestFactory) {
return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
}
// 如果没有 RestTemplateCustomizer 实例才会创建
// 这里就就会为咱们所有的 restTemplate 实例添加 loadBalancerInterceptor 拦截器
@Bean
@ConditionalOnMissingBean
public RestTemplateCustomizer restTemplateCustomizer(
final LoadBalancerInterceptor loadBalancerInterceptor) {
return restTemplate -> {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
};
}
}
// ...
}
针对下面的代码我们可以总结一下：
如果需要使用负载均衡，工程下面必须要有 RestTemplate 类， 然后Spring 容器中要有 LoadBalancerClient 的实例。
LoadBalancerClient 在 spring-cloud-netflix-ribbon 中只有一个实现类: RibbonLoadBalancerClient
利用 Spring 的 SmartInitializingSingleton 拓展点，在 restTemplateCustomizer() 中为所有的 RestTemplate 添加 LoadBalancerInterceptor 拦截器
其实 LoadBalancer 的本质就是通过拦截器。利用 RestTemplate 的拓展点来实现请求服务的负载均衡。
LoadBalancerInterceptor
LoadBalancerInterceptor 拦截器会将请求交给 LoadBalancerClient 去处理，首先会选择一个 ILoadBalancer 的实现来处理获取和选择服务，然后通过 serviceName 和负载均衡算法去选择 Server 对象。最后执行请求。
代码语言：javascript复制public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
// 负载均衡
private LoadBalancerClient loadBalancer;
// 构建请求
private LoadBalancerRequestFactory requestFactory;
// ...
@Override
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
final ClientHttpRequestExecution execution) throws IOException {
final URI originalUri = request.getURI();
String serviceName = originalUri.getHost();
return this.loadBalancer.execute(serviceName,
this.requestFactory.createRequest(request, body, execution));
}
}
RibbonLoadBalancerClient
我们通过跟踪 this.loadBalancer.execute 代码发现。最终所有的请求都交由 RibbonLoadBalancerClient 去处理。它实现了。LoadBalancerClient 接口， 代码如下：
代码语言：javascript复制public interface ServiceInstanceChooser {
// 通过 serviceId 选择具体的服务实例
ServiceInstance choose(String serviceId);
}
public interface LoadBalancerClient extends ServiceInstanceChooser {
<T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;
<T> T execute(String serviceId, ServiceInstance serviceInstance,
LoadBalancerRequest<T> request) throws IOException;
// 将服务实例信息替换还具体的 IP 信息
URI reconstructURI(ServiceInstance instance, URI original);
}
我们先来分析 RibbonLoadBalancerClient 的 choose 方法
代码语言：javascript复制@Override
public ServiceInstance choose(String serviceId) {
return choose(serviceId, null);
}
// 通过服务名选择具体的服务实例
public ServiceInstance choose(String serviceId, Object hint) {
Server server = getServer(getLoadBalancer(serviceId), hint);
if (server == null) {
return null;
}
return new RibbonServer(serviceId, server, isSecure(server, serviceId),
serverIntrospector(serviceId).getMetadata(server));
}
// 通过服务名选择一个负载均衡器, 默认是 `ZoneAwareLoadBalancer`
protected ILoadBalancer getLoadBalancer(String serviceId) {
return this.clientFactory.getLoadBalancer(serviceId);
}
// 获取服务
protected Server getServer(ILoadBalancer loadBalancer) {
return getServer(loadBalancer, null);
}
protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
if (loadBalancer == null) {
return null;
}
// Use 'default' on a null hint, or just pass it on?
return loadBalancer.chooseServer(hint != null ? hint : "default");
}
LoadBalancerInterceptor 执行的时候是直接委托执行的 loadBalancer.execute() 这个方法：
代码语言：javascript复制// LoadBalancerRequest 是通过 LoadBalancerRequestFactory.createRequest(request, body, execution) 创建
// 它实现 LoadBalancerRequest 接口是用的一个匿名内部类，泛型类型是ClientHttpResponse
// 因为最终执行的显然还是执行器：ClientHttpRequestExecution.execute()
@Override
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
return execute(serviceId, request, null);
}
public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint) throws IOException {
// 拿到负载均衡器，然后拿到一个serverInstance实例
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
Server server = getServer(loadBalancer, hint);
if (server == null) { // 若没找到就直接抛出异常。这里使用的是IllegalStateException这个异常
throw new IllegalStateException("No instances available for " serviceId);
}
// 把Server适配为RibbonServer isSecure：客户端是否安全
// serverIntrospector内省 参考配置文件：ServerIntrospectorProperties
RibbonServer ribbonServer = new RibbonServer(serviceId, server,
isSecure(server, serviceId), serverIntrospector(serviceId).getMetadata(server));
//调用本类的重载接口方法
return execute(serviceId, ribbonServer, request);
}
// 它的参数是 ServiceInstance --> 已经确定了唯一的Server实例
@Override
public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {
// 拿到 Server，RibbonServer 是 execute 时的唯一实现
Server server = null;
if (serviceInstance instanceof RibbonServer) {
server = ((RibbonServer) serviceInstance).getServer();
}
if (server == null) {
throw new IllegalStateException("No instances available for " serviceId);
}
// 执行的上下文是和serviceId绑定的
RibbonLoadBalancerContext context = this.clientFactory.getLoadBalancerContext(serviceId);
...
// 真正的向server发送请求，得到返回值
// 因为有拦截器，所以这里肯定说执行的是InterceptingRequestExecution.execute()方法
// so会调用ServiceRequestWrapper.getURI()，从而就会调用reconstructURI()方法
T returnVal = request.apply(serviceInstance);
return returnVal;
... // 异常处理
}
returnVal 是一个 ClientHttpResponse，最后交给 handleResponse()方法来处理异常情况（若存在的话），若无异常就交给提取器提值：responseExtractor.extractData(response)，这样整个请求就算全部完成了。
ZoneAwareLoadBalancer
负载均衡器 ZoneAwareLoadBalancer 的类图结构如下图所示。它 DynamicServerListLoadBalancer 它的父类， 核心方法 重置和初始化：restOfInit(clientConfig) 更新服务列表：updateListOfServers(); 这个方需要调用到 ServerList.getUpdatedListOfServers() 这里就会调用到具体的注册中心实现，以 Nacos 为例他的实现就是 NacosServerList#getUpdatedListOfServers();
- 更新所有服务列表：updateAllServerList();
- 设置所有服务列表 setServersList() ZoneAwareLoadBalancer 它的核心方法：
- 选择服务实例 chooseServer()
- 选择负载均衡器 getLoadBalancer
- 选择区域内的服务实例：zoneLoadBalancer.chooseServer
Ribbon 总结
针对 @LoadBalanced 下的 RestTemplate 的使用，我总结如下：
- 传入的String类型的url必须是绝对路径（http://...），否则抛出异常：java.lang.IllegalArgumentException: URI is not absolute
- serviceId 不区分大小写（http://order-service/...效果同http://OERDER-SERVICE/...）
- serviceId 后请不要跟 port 端口号
最后，需要特别指出的是：标注有@LoadBalanced 的 RestTemplate 只能填写 serviceId 而不能再写 IP地址/域名去发送请求了， 若你的项目中两种 case 都有需要，需要定义多个 RestTemplate 分别应对不同的使用场景
Nacos 服务查询
客户端查询
如果我们使用默认的 Nacos 客户端，那么走的就是 NacosServerList#getUpdatedListOfServers();接口来查询服务列表。
代码语言：javascript复制public class NacosServerList extends AbstractServerList<NacosServer> {
private NacosDiscoveryProperties discoveryProperties;
@Override
public List<NacosServer> getUpdatedListOfServers() {
return getServers();
}
private List<NacosServer> getServers() {
try {
String group = discoveryProperties.getGroup();
// discoveryProperties.namingServiceInstance()
// 最终通过反射获取 com.alibaba.nacos.client.naming.NacosNamingService 实例
List<Instance> instances = discoveryProperties.namingServiceInstance()
.selectInstances(serviceId, group, true);
return instancesToServerList(instances);
}
catch (Exception e) {
throw new IllegalStateException(
"Can not get service instances from nacos, serviceId=" serviceId,
e);
}
}
}
然后调用 selectInstances 方法
代码语言：javascript复制@Override
public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy,
boolean subscribe) throws NacosException {
ServiceInfo serviceInfo;
// subscribe 默认传的是 true
if (subscribe) {
serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
StringUtils.join(clusters, ","));
} else {
serviceInfo = hostReactor
.getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
StringUtils.join(clusters, ","));
}
return selectInstances(serviceInfo, healthy);
}
其实核心的逻辑在 hostReactor.getServiceInfo 在查询服务信息里面会把当前的 serviceName、 clusters 转换为 key， 然后通过 getServiceInfo0 方法查询服务信息这里主要是查询的是本地的数据。
如果 null == serviceObj 会在 updateServiceNow 里面去调用 /instance/list接口查询服务信息
代码语言：javascript复制public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
NAMING_LOGGER.debug("failover-mode: " failoverReactor.isFailoverSwitch());
String key = ServiceInfo.getKey(serviceName, clusters);
if (failoverReactor.isFailoverSwitch()) {
return failoverReactor.getService(key);
}
ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
if (null == serviceObj) {
serviceObj = new ServiceInfo(serviceName, clusters);
serviceInfoMap.put(serviceObj.getKey(), serviceObj);
updatingMap.put(serviceName, new Object());
updateServiceNow(serviceName, clusters);
updatingMap.remove(serviceName);
} else if (updatingMap.containsKey(serviceName)) {
// UPDATE_HOLD_INTERVAL 为常量默认金辉进去
if (UPDATE_HOLD_INTERVAL > 0) {
// hold a moment waiting for update finish
synchronized (serviceObj) {
try {
// 最大等待时间 5s, 在更新 serviceObj 之后, 就会执行 notifyAll()
// 方法入口 updateService(String serviceName, String clusters)
// 最大延迟 2s DEFAULT_DELAY = 1
serviceObj.wait(UPDATE_HOLD_INTERVAL);
} catch (InterruptedException e) {
NAMING_LOGGER
.error("[getServiceInfo] serviceName:" serviceName ", clusters:" clusters, e);
}
}
}
}
// 通过 Schedule 更新 服务信息
scheduleUpdateIfAbsent(serviceName, clusters);
// 获取最新的值
return serviceInfoMap.get(serviceObj.getKey());
}
代码看到这里我们不难理解，为什么第一次 Ribbon 调用的时候都会比较慢，因为它回去初始化服务列表，然后通过 Nacos Client 去 Nacos 查询服务实例信息。
服务端处理
服务端通过 /instance/list 接口来处理服务实例信息查询请求。首先服务实例信息都是被存储在 ConcurrentHashMap 中
代码语言：javascript复制 /**
* Map(namespace, Map(group::serviceName, Service)).
*/
private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
在我们查询的过程中主要是通过 ServiceManager 来进行管理， 核心的入口方法在 InstanceController#doSrvIpxt 中
代码语言：javascript复制 public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
ClientInfo clientInfo = new ClientInfo(agent);
ObjectNode result = JacksonUtils.createEmptyJsonNode();
Service service = serviceManager.getService(namespaceId, serviceName);
long cacheMillis = switchDomain.getDefaultCacheMillis();
// now try to enable the push
try {
if (udpPort > 0 && pushService.canEnablePush(agent)) {
pushService
.addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),
pushDataSource, tid, app);
cacheMillis = switchDomain.getPushCacheMillis(serviceName);
}
} catch (Exception e) {
Loggers.SRV_LOG
.error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);
cacheMillis = switchDomain.getDefaultCacheMillis();
}
if (service == null) {
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
}
result.put("name", serviceName);
result.put("clusters", clusters);
result.put("cacheMillis", cacheMillis);
result.replace("hosts", JacksonUtils.createEmptyArrayNode());
return result;
}
checkIfDisabled(service);
List<Instance> srvedIPs;
// 查询所有的服务
// 内部会更新服务列表
// allInstances.addAll(persistentInstances);
// allInstances.addAll(ephemeralInstances);
srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));
// filter ips using selector:
if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) {
srvedIPs = service.getSelector().select(clientIP, srvedIPs);
}
if (CollectionUtils.isEmpty(srvedIPs)) {
if (Loggers.SRV_LOG.isDebugEnabled()) {
Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
}
if (clientInfo.type == ClientInfo.ClientType.JAVA
&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
result.put("dom", serviceName);
} else {
result.put("dom", NamingUtils.getServiceName(serviceName));
}
result.put("name", serviceName);
result.put("cacheMillis", cacheMillis);
result.put("lastRefTime", System.currentTimeMillis());
result.put("checksum", service.getChecksum());
result.put("useSpecifiedURL", false);
result.put("clusters", clusters);
result.put("env", env);
result.set("hosts", JacksonUtils.createEmptyArrayNode());
result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
return result;
}
Map<Boolean, List<Instance>> ipMap = new HashMap<>(2);
ipMap.put(Boolean.TRUE, new ArrayList<>());
ipMap.put(Boolean.FALSE, new ArrayList<>());
for (Instance ip : srvedIPs) {
ipMap.get(ip.isHealthy()).add(ip);
}
if (isCheck) {
result.put("reachProtectThreshold", false);
}
double threshold = service.getProtectThreshold();
if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) {
Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);
if (isCheck) {
result.put("reachProtectThreshold", true);
}
ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));
ipMap.get(Boolean.FALSE).clear();
}
if (isCheck) {
result.put("protectThreshold", service.getProtectThreshold());
result.put("reachLocalSiteCallThreshold", false);
return JacksonUtils.createEmptyJsonNode();
}
ArrayNode hosts = JacksonUtils.createEmptyArrayNode();
for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) {
List<Instance> ips = entry.getValue();
if (healthyOnly && !entry.getKey()) {
continue;
}
for (Instance instance : ips) {
// remove disabled instance:
if (!instance.isEnabled()) {
continue;
}
ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();
ipObj.put("ip", instance.getIp());
ipObj.put("port", instance.getPort());
// deprecated since nacos 1.0.0:
ipObj.put("valid", entry.getKey());
ipObj.put("healthy", entry.getKey());
ipObj.put("marked", instance.isMarked());
ipObj.put("instanceId", instance.getInstanceId());
ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata()));
ipObj.put("enabled", instance.isEnabled());
ipObj.put("weight", instance.getWeight());
ipObj.put("clusterName", instance.getClusterName());
if (clientInfo.type == ClientInfo.ClientType.JAVA
&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
ipObj.put("serviceName", instance.getServiceName());
} else {
ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName()));
}
ipObj.put("ephemeral", instance.isEphemeral());
hosts.add(ipObj);
}
}
result.replace("hosts", hosts);
if (clientInfo.type == ClientInfo.ClientType.JAVA
&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
result.put("dom", serviceName);
} else {
result.put("dom", NamingUtils.getServiceName(serviceName));
}
result.put("name", serviceName);
result.put("cacheMillis", cacheMillis);
result.put("lastRefTime", System.currentTimeMillis());
result.put("checksum", service.getChecksum());
result.put("useSpecifiedURL", false);
result.put("clusters", clusters);
result.put("env", env);
result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
return result;
}
在上面的核心逻辑主要是：
- 调用 service.srvIPs 方法查询所有的服务实例信息
- Cluster#allIPs会将所有的服务注册信息写到服务注册列表。
参考链接
- https://nacos.io
- https://zhuanlan.zhihu.com
- https://blog.csdn.net/f641385712/article/details/100788040
