10、Nacos服务订阅服务端源码分析

本文收录于专栏 Nacos
推荐阅读：Nacos 架构 & 原理

文章目录

前言
一、RequestHandlerRegistry
二、SubscribeServiceRequestHandler
三、EphemeralClientOperationServiceImpl
四、ClientServiceIndexesManager
总结

前言

本篇开始学习Nacos中服务订阅相关的源码

一、RequestHandlerRegistry

再前边看客户端注册到Nacos时，我们讲过一部分grpc的逻辑。GrpcRequestAcceptor会接收grpc请求，统一处理，源码如下：

@Autowired
RequestHandlerRegistry requestHandlerRegistry;
...
RequestHandler requestHandler = requestHandlerRegistry.getByRequestType(type);
...
Response response = requestHandler.handleRequest(request, requestMeta);

可以看到RequestHandlerRegistry根据不同的请求类型获取了相应的RequestHandler去处理请求，那么我们看下这个类的源码是如何组装请求类型和请求处理器RequestHandler的对应关系的。

@Service
public class RequestHandlerRegistry implements ApplicationListener<ContextRefreshedEvent> {
    
    Map<String, RequestHandler> registryHandlers = new HashMap<>();
	...

	@Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        Map<String, RequestHandler> beansOfType = event.getApplicationContext().getBeansOfType(RequestHandler.class);
        ...
        //将处理器保存到本地缓存
        registryHandlers.putIfAbsent(tClass.getSimpleName(), requestHandler);
        }
    }
}

可以看到，RequestHandlerRegistry实现了ApplicationListener，监听了spring的ContextRefreshedEvent事件，在处理事件时将所有RequestHandler的实现保存到本地缓存registryHandlers中。

二、SubscribeServiceRequestHandler

@Override
@Secured(action = ActionTypes.READ)
public SubscribeServiceResponse handle(SubscribeServiceRequest request, RequestMeta meta) throws NacosException {
    String namespaceId = request.getNamespace();
    String serviceName = request.getServiceName();
    String groupName = request.getGroupName();
    String app = request.getHeader("app", "unknown");
    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
    Service service = Service.newService(namespaceId, groupName, serviceName, true);
    Subscriber subscriber = new Subscriber(meta.getClientIp(), meta.getClientVersion(), app, meta.getClientIp(),
            namespaceId, groupedServiceName, 0, request.getClusters());
    ServiceInfo serviceInfo = ServiceUtil.selectInstancesWithHealthyProtection(serviceStorage.getData(service),
            metadataManager.getServiceMetadata(service).orElse(null), subscriber.getCluster(), false,
            true, subscriber.getIp());
    if (request.isSubscribe()) {
    	//订阅
        clientOperationService.subscribeService(service, subscriber, meta.getConnectionId());
        NotifyCenter.publishEvent(new SubscribeServiceTraceEvent(System.currentTimeMillis(),
                meta.getClientIp(), service.getNamespace(), service.getGroup(), service.getName()));
    } else {
    	//退订
        clientOperationService.unsubscribeService(service, subscriber, meta.getConnectionId());
        NotifyCenter.publishEvent(new UnsubscribeServiceTraceEvent(System.currentTimeMillis(),
                meta.getClientIp(), service.getNamespace(), service.getGroup(), service.getName()));
    }
    return new SubscribeServiceResponse(ResponseCode.SUCCESS.getCode(), "success", serviceInfo);
}

我们从中发现一个熟悉的类EphemeralClientOperationServiceImpl clientOperationService，这个类目前除了客户端注册之外还处理了客户端订阅。

三、EphemeralClientOperationServiceImpl

@Override
public void subscribeService(Service service, Subscriber subscriber, String clientId) {
    Service singleton = ServiceManager.getInstance().getSingletonIfExist(service).orElse(service);
    Client client = clientManager.getClient(clientId);
    if (!clientIsLegal(client, clientId)) {
        return;
    }
    client.addServiceSubscriber(singleton, subscriber);
    client.setLastUpdatedTime();
    NotifyCenter.publishEvent(new ClientOperationEvent.ClientSubscribeServiceEvent(singleton, clientId));
}

这里的处理逻辑如下：

根据clientId去ClientManager中获取客户端对象Client
- 如果可以获取客户端对象，则说明当前客户端已经处理过订阅逻辑，直接返回
发布客户端订阅事件ClientOperationEvent.ClientSubscribeServiceEvent

四、ClientServiceIndexesManager

private void handleClientOperation(ClientOperationEvent event) {
    Service service = event.getService();
    String clientId = event.getClientId();
    if (event instanceof ClientOperationEvent.ClientRegisterServiceEvent) {
        //处理客户端注册事件
        addPublisherIndexes(service, clientId);
    } else if (event instanceof ClientOperationEvent.ClientDeregisterServiceEvent) {
        removePublisherIndexes(service, clientId);
    } else if (event instanceof ClientOperationEvent.ClientSubscribeServiceEvent) {
        //处理客户单订阅事件
        addSubscriberIndexes(service, clientId);
    } else if (event instanceof ClientOperationEvent.ClientUnsubscribeServiceEvent) {
        removeSubscriberIndexes(service, clientId);
    }
}

查找客户单订阅事件ClientOperationEvent.ClientSubscribeServiceEvent的处理类，我们再次看到了熟悉的类ClientServiceIndexesManager。看看addSubscriberIndexes是如何处理客户端订阅事件的：

private final ConcurrentMap<Service, Set<String>> subscriberIndexes = new ConcurrentHashMap<>();

private void addSubscriberIndexes(Service service, String clientId) {
    subscriberIndexes.computeIfAbsent(service, key -> new ConcurrentHashSet<>());
    // Fix #5404, Only first time add need notify event.
    if (subscriberIndexes.get(service).add(clientId)) {
        NotifyCenter.publishEvent(new ServiceEvent.ServiceSubscribedEvent(service, clientId));
    }
}