Nacos源码系列之服务发现（二）

文章目录

一、前言
二、服务端-服务注册
三、客户端-服务发现
- （1）注册listener
- （2）subscribe
四、服务端-服务发现
五、小结

一、前言

上篇博客介绍了客户端服务注册的流程，本篇介绍服务端的服务注册，服务发现等核心流程。

二、服务端-服务注册

入口在InstanceController的register()，核心逻辑在ServiceManager类中。

ServiceManager：核心服务管理类，管理服务、实例信息。包含nacos的服务注册表。

//Register an instance to a service in AP mode.
//先创建service，校验service，之后创建instance
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {createEmptyService(namespaceId, serviceName, instance.isEphemeral());Service service = getService(namespaceId, serviceName);checkServiceIsNull(service, namespaceId, serviceName);addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}public void createEmptyService(String namespaceId, String serviceName, boolean local) throws NacosException {createServiceIfAbsent(namespaceId, serviceName, local, null);
}public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)throws NacosException {Service service = getService(namespaceId, serviceName);if (service == null) {Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);service = new Service();service.setName(serviceName);service.setNamespaceId(namespaceId);service.setGroupName(NamingUtils.getGroupName(serviceName));// now validate the service. if failed, exception will be thrownservice.setLastModifiedMillis(System.currentTimeMillis());service.recalculateChecksum();if (cluster != null) {cluster.setService(service);service.getClusterMap().put(cluster.getName(), cluster);}service.validate();putServiceAndInit(service);if (!local) {addOrReplaceService(service);}}
}//添加instance，获取service后，使用synchronized锁住service，防止并发操作。
public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)throws NacosException {String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);Service service = getService(namespaceId, serviceName);synchronized (service) {List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);Instances instances = new Instances();instances.setInstanceList(instanceList);consistencyService.put(key, instances);}
}

ServiceManager的init()还做了一些初始化工作：


@PostConstruct
public void init() {//service reporter的定时任务执行器GlobalExecutor.scheduleServiceReporter(new ServiceReporter(), 60000, TimeUnit.MILLISECONDS);//服务更新管理的定时任务执行器GlobalExecutor.submitServiceUpdateManager(new UpdatedServiceProcessor());if (emptyServiceAutoClean) {Loggers.SRV_LOG.info("open empty service auto clean job, initialDelay : {} ms, period : {} ms",cleanEmptyServiceDelay, cleanEmptyServicePeriod);// delay 60s, period 20s;// This task is not recommended to be performed frequently in order to avoid// the possibility that the service cache information may just be deleted// and then created due to the heartbeat mechanism//自动清理空的服务的执行器GlobalExecutor.scheduleServiceAutoClean(new EmptyServiceAutoCleaner(this, distroMapper), cleanEmptyServiceDelay,cleanEmptyServicePeriod);}try {Loggers.SRV_LOG.info("listen for service meta change");//监听meta keyconsistencyService.listen(KeyBuilder.SERVICE_META_KEY_PREFIX, this);} catch (NacosException e) {Loggers.SRV_LOG.error("listen for service meta change failed!");}
}

注册表：

双重map结构


/**
* Map(namespace, Map(group::serviceName, Service)).
*/
private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();

三、客户端-服务发现

客户端的服务发现，可以先从源码中的例子入手。例子位于nacos-example中的NamingExample。


NamingService naming = NamingFactory.createNamingService(properties);
naming.subscribe("nacos.test.3", new AbstractEventListener() {//EventListener onEvent is sync to handle, If process too low in onEvent, maybe block other onEvent callback.//So you can override getExecutor() to async handle event.@Overridepublic Executor getExecutor() {return executor;}@Overridepublic void onEvent(Event event) {System.out.println("serviceName: " + ((NamingEvent) event).getServiceName());System.out.println("instances from event: " + ((NamingEvent) event).getInstances());}
});

进入NacosNamingService，注册listener，订阅。


public void subscribe(String serviceName, String groupName, List<String> clusters, EventListener listener)throws NacosException {if (null == listener) {return;}String clusterString = StringUtils.join(clusters, ",");changeNotifier.registerListener(groupName, serviceName, clusterString, listener);clientProxy.subscribe(serviceName, groupName, clusterString);
}

（1）注册listener

InstancesChangeNotifier


public class InstancesChangeNotifier extends Subscriber<InstancesChangeEvent> {//缓存map，用来存储listenerprivate final Map<String, ConcurrentHashSet<EventListener>> listenerMap = new ConcurrentHashMap<String, ConcurrentHashSet<EventListener>>();//Object锁private final Object lock = new Object();//注册listener，根据key在map中查询对应的listener set。使用synchronized锁定防止并发操作。
//若查到，在set中添加本次要添加的listener。
//若没有对应的set，新建空set就添加到map中，在set中加入listener。
public void registerListener(String groupName, String serviceName, String clusters, EventListener listener) {String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), clusters);ConcurrentHashSet<EventListener> eventListeners = listenerMap.get(key);if (eventListeners == null) {synchronized (lock) {eventListeners = listenerMap.get(key);if (eventListeners == null) {eventListeners = new ConcurrentHashSet<EventListener>();listenerMap.put(key, eventListeners);}}}eventListeners.add(listener);
}
}

（2）subscribe

NamingClientProxy

对于NamingClientProxy接口，实现类主要有这三个。上篇提到过，这里不再赘述，直接说具体业务逻辑。

NamingClientProxyDelegate


public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {NAMING_LOGGER.info("[SUBSCRIBE-SERVICE] service:{}, group:{}, clusters:{} ", serviceName, groupName, clusters);String serviceNameWithGroup = NamingUtils.getGroupedName(serviceName, groupName);String serviceKey = ServiceInfo.getKey(serviceNameWithGroup, clusters);
//如果存在update service task，则执行serviceInfoUpdateService.scheduleUpdateIfAbsent(serviceName, groupName, clusters);
//校验：若sserviceInfoHolder中没有该ervice或未被订阅，就开始订阅ServiceInfo result = serviceInfoHolder.getServiceInfoMap().get(serviceKey);if (null == result || !isSubscribed(serviceName, groupName, clusters)) {result = grpcClientProxy.subscribe(serviceName, groupName, clusters);}
//更新缓存中的serviceserviceInfoHolder.processServiceInfo(result);return result;
}public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {String serviceKey = serviceInfo.getKey();if (serviceKey == null) {return null;}ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());//若service为空或者error，则忽略if (isEmptyOrErrorPush(serviceInfo)) {return oldService;}serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
//校验新旧service，是否发生变化。boolean changed = isChangedServiceInfo(oldService, serviceInfo);if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));}MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
//若服务发生变化，发布事件通知。if (changed) {NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceInfo.getKey(),JacksonUtils.toJson(serviceInfo.getHosts()));NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(),serviceInfo.getClusters(), serviceInfo.getHosts()));DiskCache.write(serviceInfo, cacheDir);}return serviceInfo;
}

NamingGrpcClientProxy


public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {if (NAMING_LOGGER.isDebugEnabled()) {NAMING_LOGGER.debug("[GRPC-SUBSCRIBE] service:{}, group:{}, cluster:{} ", serviceName, groupName, clusters);}
//放入缓存redoService.cacheSubscriberForRedo(serviceName, groupName, clusters);return doSubscribe(serviceName, groupName, clusters);
}private final ConcurrentMap<String, SubscriberRedoData> subscribes = new ConcurrentHashMap<>();public void cacheSubscriberForRedo(String serviceName, String groupName, String cluster) {String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);SubscriberRedoData redoData = SubscriberRedoData.build(serviceName, groupName, cluster);synchronized (subscribes) {subscribes.put(key, redoData);}
}//订阅，构造request发起grpc  其中requestToServer() 上篇提到过，这里不做赘述。
public ServiceInfo doSubscribe(String serviceName, String groupName, String clusters) throws NacosException {SubscribeServiceRequest request = new SubscribeServiceRequest(namespaceId, groupName, serviceName, clusters,true);SubscribeServiceResponse response = requestToServer(request, SubscribeServiceResponse.class);redoService.subscriberRegistered(serviceName, groupName, clusters);return response.getServiceInfo();
}public void subscriberRegistered(String serviceName, String groupName, String cluster) {String key = ServiceInfo.getKey(NamingUtils.getGroupedName(serviceName, groupName), cluster);synchronized (subscribes) {SubscriberRedoData redoData = subscribes.get(key);if (null != redoData) {redoData.setRegistered(true);}}
}

NamingHttpClientProxy

public ServiceInfo subscribe(String serviceName, String groupName, String clusters) throws NacosException {return queryInstancesOfService(serviceName, groupName, clusters, pushReceiver.getUdpPort(), false);
}public ServiceInfo queryInstancesOfService(String serviceName, String groupName, String clusters, int udpPort,boolean healthyOnly) throws NacosException {final Map<String, String> params = new HashMap<String, String>(16);params.put(CommonParams.NAMESPACE_ID, namespaceId);params.put(CommonParams.SERVICE_NAME, NamingUtils.getGroupedName(serviceName, groupName));params.put(CLUSTERS_PARAM, clusters);params.put(UDP_PORT_PARAM, String.valueOf(udpPort));params.put(CLIENT_IP_PARAM, NetUtils.localIP());params.put(HEALTHY_ONLY_PARAM, String.valueOf(healthyOnly));String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);if (StringUtils.isNotEmpty(result)) {return JacksonUtils.toObj(result, ServiceInfo.class);}return new ServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), clusters);
}

四、服务端-服务发现

客户端服务发现调用的接口在nacos-naming的InstanceController和InstanceControllerV2中。V2中也兼容了V1的逻辑。

    @GetMapping("/list")@Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)public Object list(HttpServletRequest request) throws Exception {String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);NamingUtils.checkServiceNameFormat(serviceName);String agent = WebUtils.getUserAgent(request);String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));String app = WebUtils.optional(request, "app", StringUtils.EMPTY);String env = WebUtils.optional(request, "env", StringUtils.EMPTY);String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);Subscriber subscriber = new Subscriber(clientIP + ":" + udpPort, agent, app, clientIP, namespaceId, serviceName,udpPort, clusters);return getInstanceOperator().listInstance(namespaceId, serviceName, subscriber, clusters, healthyOnly);}public ServiceInfo listInstance(String namespaceId, String serviceName, Subscriber subscriber, String cluster,  boolean healthOnly) throws Exception {...String clientIP = subscriber.getIp();ServiceInfo result = new ServiceInfo(serviceName, cluster);//具体见下面...// now try to enable the pushtry {if (subscriber.getPort() > 0 && pushService.canEnablePush(subscriber.getAgent())) {subscriberServiceV1.addClient(namespaceId, serviceName, cluster, subscriber.getAgent(),new InetSocketAddress(clientIP, subscriber.getPort()), pushDataSource, StringUtils.EMPTY,StringUtils.EMPTY);cacheMillis = switchDomain.getPushCacheMillis(serviceName);}} catch (Exception e) {Loggers.SRV_LOG.error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP,subscriber.getPort(), e);cacheMillis = switchDomain.getDefaultCacheMillis();}...return result;}

Nacos Client并不是完全依赖定时任务来感知Service的变化，为了尽量的去弥补这个延迟问题，采用一个UDP的变更通知设计，客户端调用/nacos/v1/ns/instance/list接口的时候会传入一个UDP的port，在接口中会把Service订阅的其他Service加入到一个com.alibaba.nacos.naming.push.PushService#clientMap中去，如果Service中的Instance发生了变化，取出订阅了此实例的客户端列表，并通过UDP的方式进行通知。

五、小结

在naming service模块：
1、nacos-client
主要是nacos客户端的业务逻辑，与服务端的调用基本有两种方式，grpc和http，分别对应NamingGrpcClientProxy和NamingHttpClientProxy。

2、nacos-naming
nacos naming的服务端，提供对外api接口，包括对服务、实例等的查询、更新、删除等操作。

下篇计划重点介绍Grpc和UDP相关的内容。