1.InstanceController#list()
Nacos Server处理订阅请求
主要还是从请求中获取参数, 比如namespceId、serviceName、agent(指定提交请求的客户端是哪种类型)、clusters、clusterIP、udpPort(后续UDP通信会使用)、app、tenant, 最后调用方法对参数进行处理
2.InstanceController#doSrvIpxt()
对请求进行详细处理
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 {
// 不同agent，生成不同的clientInfo
ClientInfo clientInfo = new ClientInfo(agent);
// 创建一个JSON Node，其就是当前方法返回的结果。后续代码就是对这个Node的各种初始化
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)) {
// 创建当前发出订阅请求的Nacos client的UDP Client, PushClient
// 注意，在Nacos的UDP通信中，Nacos Server充当的是UDP Client，Nacos Client充当的是UDP Server
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);
// 注意，hosts为空
result.replace("hosts", JacksonUtils.createEmptyArrayNode());
return result;
}
// 代码直到这里，说明注册表中存在该服务
// 检测该服务是否被禁。若是被禁的服务，直接抛出异常
checkIfDisabled(service);
List<Instance> srvedIPs;
// 获取到当前服务的所有实例，包含所有持久/临时实例
srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));
// filter ips using selector:
// 若选择器不空，则根据选择算法选择可用的intance列表，默认情况下，选择器不做任何过滤
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);
// 注意，hosts为空
result.set("hosts", JacksonUtils.createEmptyArrayNode());
result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
return result;
}
// 代码走到这里，说明具有可用的instance
Map<Boolean, List<Instance>> ipMap = new HashMap<>(2);
// 这个map只有两个key，True与False
// key为true的value中存放的是所有健康的instance
// key为false的value存放的是所有不健康的instance
ipMap.put(Boolean.TRUE, new ArrayList<>());
ipMap.put(Boolean.FALSE, new ArrayList<>());
// 根据instance的健康状态，将所有instance分流放入map的不同key的value中
for (Instance ip : srvedIPs) {
// 这个语句写的非常好
// 健康加入健康的列表, 不健康的加入不健康的列表
ipMap.get(ip.isHealthy()).add(ip);
}
// isCheck为true，表示需要检测instance的保护阈值
if (isCheck) {
// reachProtectThreshold 是否达到了保护阈值, false 为没有达到
result.put("reachProtectThreshold", false);
}
// 获取服务的保护阈值
double threshold = service.getProtectThreshold();
// 若 "健康instance数量/instance总数" <= 保护阈值，则说明需要启动保护机制了
if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) {
Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);
if (isCheck) {
// true表示启动保护机制
result.put("reachProtectThreshold", true);
}
// 健康数量小于阈值, 则从所有实例中调用, 可能会有不健康实例, 可以保证健康实例不被压崩溃
// 将所有不健康的instance添加到的key为true的instance列表，
// 即key为true的value中（instance列表）存放的是所有instance实例
// 包含所有健康的与不健康的instance
ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));
// 清空key为false的value（不健康的instance列表）
ipMap.get(Boolean.FALSE).clear();
}
if (isCheck) {
result.put("protectThreshold", service.getProtectThreshold());
result.put("reachLocalSiteCallThreshold", false);
return JacksonUtils.createEmptyJsonNode();
}
ArrayNode hosts = JacksonUtils.createEmptyArrayNode();
// 注意，这个ipMap中存放着所有健康与不健康的instance列表
for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) {
List<Instance> ips = entry.getValue();
// 若客户端只要健康的instance，且当前遍历的map的key为false，则跳过
if (healthyOnly && !entry.getKey()) {
continue;
}
// 遍历的这个ips可能是所有不健康的instance列表，
// 也可能是所有健康的instance列表，
// 也可能是所有健康与不健康的instance列表总和
for (Instance instance : ips) {
// 跳过禁用的instance
if (!instance.isEnabled()) {
continue;
}
ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();
// 将当前遍历的instance转换为JSON
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);
} // end-for
} // end-for
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;
}
1、 不同agent，生成不同的clientInfo,java、c、c++、go、nginx、dnsf；
2、 pushService.addClient():创建当前发出订阅请求的Nacosclient的UDPClient,PushClient,NacosServer充当的是UDPClient，NacosClient充当的是UDPServer；
获取到了UDP通信客户端PushClient, 并写入到一个缓存map中
public void addClient(PushClient client) {
// client is stored by key 'serviceName' because notify event is driven by serviceName change
String serviceKey = UtilsAndCommons.assembleFullServiceName(client.getNamespaceId(), client.getServiceName());
// clientMap是一个缓存map，用于存放当前Nacos Server中所有instance对应的UDP Client
// 其是一个双层map，外层map的key为 namespaceId##groupId@@微服务名称，value为内层map
// 内层map的key为代表一个instance的字符串，value为该instance对应的UDP Client，即PushClient
ConcurrentMap<String, PushClient> clients = clientMap.get(serviceKey);
// 若当前服务的内层map为null，则创建一个并放入到缓存map
if (clients == null) {
clientMap.putIfAbsent(serviceKey, new ConcurrentHashMap<>(1024));
clients = clientMap.get(serviceKey);
}
PushClient oldClient = clients.get(client.toString());
// 从内层map中获取当前instance对应的的PushClient，
// 若该PushClient不为null，则更新一个最后引用时间戳;
// 若该PushClient为null，则将当前这个PushClient作为PushClient
// 写入到内层map，即写入到了缓存map
if (oldClient != null) {
// 更新最后引用时间戳
oldClient.refresh();
} else {
PushClient res = clients.putIfAbsent(client.toString(), client);
if (res != null) {
Loggers.PUSH.warn("client: {} already associated with key {}", res.getAddrStr(), res.toString());
}
Loggers.PUSH.debug("client: {} added for serviceName: {}", client.getAddrStr(), client.getServiceName());
}
}
1、 获取当前服务的所有实例,包括持久和临时；
public List<Instance> srvIPs(List<String> clusters) {
if (CollectionUtils.isEmpty(clusters)) {
clusters = new ArrayList<>();
clusters.addAll(clusterMap.keySet());
}
// 获取到当前服务的所有cluster中的所有instance
return allIPs(clusters);
}
public List<Instance> allIPs(List<String> clusters) {
List<Instance> result = new ArrayList<>();
for (String cluster : clusters) {
Cluster clusterObj = clusterMap.get(cluster);
if (clusterObj == null) {
continue;
}
// 将当前遍历cluster的所有instance添加到result集合
// 包含所有持久实例与临时实例
result.addAll(clusterObj.allIPs());
}
return result;
}
public List<Instance> allIPs() {
List<Instance> allInstances = new ArrayList<>();
// 持久实例
allInstances.addAll(persistentInstances);
// 临时实例
allInstances.addAll(ephemeralInstances);
return allInstances;
}
3.总结
Nacos Server处理订阅请求的主要任务:
1、 创建了NacosClient对应的UDP通信客户端PushClient,并写入一个缓存map；
2、 从注册表中获取到指定服务的所有可用的instance,并封装为Json；