本文收录于专栏 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));
}
}
同客户端注册逻辑一样,对于客户端订阅也是采取了同样的实现方式,那就是使用一个ConcurrentHashMap
来存储客户端的订阅信息。
- key:订阅的service
- value:订阅指定service的客户端集合
总结
通过以上代码我们可以看出,比起客户端注册,客户端订阅的代码在整体结构设计上更加简单。得益于将客户端逻辑在整体上设计的高内聚
,所以除了SubscribeServiceRequestHandler
这个RequestHandler
的实现之外,我们没有接触新的类。