Spring Cloud Nacos & Ribbon

Nacos 为了适应 Spring Cloud 技术栈和在 Spring Cloud Alibaba 之前的 Spring Cloud Netflix 技术栈保持兼容，所以 Nacos 需要实现和 Spring Cloud Netfilx Ribbon 的集成，用在服务发现，以及服务到服务之间的调用功能。

核心源码分析

NacosRibbonClientConfiguration

@Configuration(proxyBeanMethods = false)
@ConditionalOnRibbonNacos
public class NacosRibbonClientConfiguration {

@Autowired
private PropertiesFactory propertiesFactory;

// 实例化一个 ServerList 实例 ，该实例类为 NacosServerList 实现了 com.netflix.loadbalancer.ServerList 接口
@Bean
@ConditionalOnMissingBean
public ServerList<?> ribbonServerList(IClientConfig config,
			NacosDiscoveryProperties nacosDiscoveryProperties) {
if (this.propertiesFactory.isSet(ServerList.class, config.getClientName())) {
			ServerList serverList = this.propertiesFactory.get(ServerList.class, config,
config.getClientName());
return serverList;
		}
		NacosServerList serverList = new NacosServerList(nacosDiscoveryProperties);
		serverList.initWithNiwsConfig(config);
return serverList;
	}

@Bean
@ConditionalOnMissingBean
public NacosServerIntrospector nacosServerIntrospector() {
return new NacosServerIntrospector();
	}

}

com.netflix.loadbalancer.ServerList 提供客户端负载均衡功能

public interface ServerList<T extends Server> {

public List<T> getInitialListOfServers();

/**
     * Return updated list of servers. This is called say every 30 secs
     * (configurable) by the Loadbalancer's Ping cycle
     * 
     */
public List<T> getUpdatedListOfServers();   

}

com.alibaba.cloud.nacos.ribbon.NacosServerList

public class NacosServerList extends AbstractServerList<NacosServer> {

@Override
public List<NacosServer> getInitialListOfServers() {
return getServers();
	}

@Override
public List<NacosServer> getUpdatedListOfServers() {
return getServers();
	}

// 该类中的核心代码获取到服务实例列表
private List<NacosServer> getServers() {
try {
			String group = discoveryProperties.getGroup();
// namingServiceInstance() 获取到 NamingService 实例对象 ， NamingService 提供了使用 http 协议和 Nacos Server 端通信的功能
			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);
		}
	}
}

com.alibaba.cloud.nacos.NacosDiscoveryProperties

@ConfigurationProperties("spring.cloud.nacos.discovery")
public class NacosDiscoveryProperties {

// 创建或者获取一个已有的 NamingService 实例
public NamingService namingServiceInstance() {

if (null != namingService) {
return namingService;
		}

try {
			namingService = NacosFactory.createNamingService(getNacosProperties());
		}
catch (Exception e) {
			log.error("create naming service error!properties={},e=,", this, e);
return null;
		}
return namingService;
	}
}

com.alibaba.nacos.client.naming.NacosNamingService

@SuppressWarnings("PMD.ServiceOrDaoClassShouldEndWithImplRule")
public class NacosNamingService implements NamingService {

// 查询服务实例列表
@Override
public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy, boolean subscribe) throws NacosException {

        ServiceInfo serviceInfo;
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);
    }
}

com.alibaba.nacos.client.naming.core.HostReactor

public class HostReactor {

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)) {

if (UPDATE_HOLD_INTERVAL > 0) {
// hold a moment waiting for update finish
synchronized (serviceObj) {
try {
                        serviceObj.wait(UPDATE_HOLD_INTERVAL);
                    } catch (InterruptedException e) {
                        NAMING_LOGGER.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
                    }
                }
            }
        }

        scheduleUpdateIfAbsent(serviceName, clusters);

return serviceInfoMap.get(serviceObj.getKey());
    }

// 去 nacos server 端查询服务实列信息更新本地 jvm 内存的中服务实例信息，存储服务实例信息到本机硬盘
public void updateServiceNow(String serviceName, String clusters) {
        ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
try {

            String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUDPPort(), false);

if (StringUtils.isNotEmpty(result)) {
// 具体如何处理从 nacos server 返回的服务实列信息数据看这个函数
                processServiceJSON(result);
            }
        } catch (Exception e) {
            NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
        } finally {
if (oldService != null) {
synchronized (oldService) {
                    oldService.notifyAll();
                }
            }
        }
    }

// 开启异步从 nacas server 查询更新本地 jvm 内存中的服务实例信息任务
public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
return;
        }

synchronized (futureMap) {
if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
return;
            }

            ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
            futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
        }
    }
}

com.alibaba.nacos.client.naming.core.HostReactor.UpdateTask

public class UpdateTask implements Runnable {
        long lastRefTime = Long.MAX_VALUE;
private String clusters;
private String serviceName;

public UpdateTask(String serviceName, String clusters) {
this.serviceName = serviceName;
this.clusters = clusters;
        }

@Override
public void run() {
try {
                ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));

if (serviceObj == null) {
                    updateServiceNow(serviceName, clusters);
                    executor.schedule(this, DEFAULT_DELAY, TimeUnit.MILLISECONDS);
return;
                }

if (serviceObj.getLastRefTime() <= lastRefTime) {
                    updateServiceNow(serviceName, clusters);
                    serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
                } else {
// if serviceName already updated by push, we should not override it
// since the push data may be different from pull through force push
                    refreshOnly(serviceName, clusters);
                }

                lastRefTime = serviceObj.getLastRefTime();

if (!eventDispatcher.isSubscribed(serviceName, clusters) &&
                    !futureMap.containsKey(ServiceInfo.getKey(serviceName, clusters))) {
// abort the update task:
                    NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);
return;
                }

                executor.schedule(this, serviceObj.getCacheMillis(), TimeUnit.MILLISECONDS);


            } catch (Throwable e) {
                NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
            }

        }