當(dāng)一個client啟動的時候， 客戶端流程線
1.入口：AbstractApplicationContext refresh 方案
image.png
2.監(jiān)聽機制實現(xiàn)服務(wù)暴露
image.png
3.namingService.registerInstance ： serviceName 服務(wù)名 instance 實例信息
image.png
NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", namespaceId, serviceName, instance); final Map<String, String> params = new HashMap<String, String>(8); params.put(Constants.REQUEST_PARAM_NAMESPACE_ID, namespaceId); params.put("ip", instance.getIp()); params.put("port", String.valueOf(instance.getPort())); params.put("weight", String.valueOf(instance.getWeight())); params.put("enable", String.valueOf(instance.isEnabled())); params.put("healthy", String.valueOf(instance.isHealthy())); params.put("metadata", JSON.toJSONString(instance.getMetadata())); params.put("serviceName", serviceName); params.put("clusterName", instance.getClusterName()); reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, HttpMethod.POST); 調(diào)用api接口- public String reqAPI(String api, Map<String, String> params, List<String> servers) {
return reqAPI(api, params, servers, HttpMethod.GET);
}
public String reqAPI(String api, Map<String, String> params, List<String> servers, String method) {
params.put(Constants.REQUEST_PARAM_NAMESPACE_ID, getNamespaceId());
if (CollectionUtils.isEmpty(servers) && StringUtils.isEmpty(nacosDomain)) {
throw new IllegalArgumentException("no server available");
}
if (servers != null && !servers.isEmpty()) { 如果服務(wù)地址不為空
Random random = new Random(System.currentTimeMillis()); 隨便獲取一個節(jié)點
int index = random.nextInt(servers.size());
for (int i = 0; i < servers.size(); i++) {
String server = servers.get(index);
try {
return callServer(api, params, server, method); 執(zhí)行調(diào)動服務(wù)
} catch (Exception e) {
NAMING_LOGGER.error("[NA] req api:" + api + " failed, server(" + server, e);
}
index = (index + 1) % servers.size(); 輪詢方案
}
throw new IllegalStateException("failed to req API:" + api + " after all servers(" + servers + ") tried");
}
for (int i = 0; i < UtilAndComs.REQUEST_DOMAIN_RETRY_COUNT; i++) { 如果超出了重試次數(shù) 報錯
try {
return callServer(api, params, nacosDomain);
} catch (Exception e) {
NAMING_LOGGER.error("[NA] req api:" + api + " failed, server(" + nacosDomain, e);
}
}
throw new IllegalStateException("failed to req API:/api/" + api + " after all servers(" + servers + ") tried");
}
- 發(fā)現(xiàn)調(diào)用遠(yuǎn)程的接口地址是 http://10.10.0.126:8848/nacos/v1/ns/instance/list
image.png
服務(wù)端
/**
* Register new instance.
*
* @param request http request
* @return 'ok' if success
* @throws Exception any error during register
*/
@CanDistro
@PostMapping
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public String register(HttpServletRequest request) throws Exception {
final String namespaceId = WebUtils
.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
NamingUtils.checkServiceNameFormat(serviceName);
final Instance instance = parseInstance(request);
getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
return "ok";
}
2. 進(jìn)入ServerManager registerInstance方法
/**
* Register an instance to a service in AP mode.
*
* <p>This method creates service or cluster silently if they don't exist.
*
* @param namespaceId id of namespace
* @param serviceName service name
* @param instance instance to register
* @throws Exception any error occurred in the process
*/
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
//創(chuàng)建一個空服務(wù)，在Nacos控制臺服務(wù)列表展示的服務(wù)信息，實際上是初始化一個serviceMap，它 是一個ConcurrentHashMap集合
createEmptyService(namespaceId, serviceName, instance.isEphemeral());
//從serviceMap中，根據(jù)namespaceId和serviceName得到一個服務(wù)對象
Service service = getService(namespaceId, serviceName);
checkServiceIsNull(service, namespaceId, serviceName);
//調(diào)用addInstance創(chuàng)建一個服務(wù)實例
addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}
3
public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
throws NacosException {
//從serviceMap中獲取服務(wù)對象
其中serviceMap是一個ConcurrentHashMap ， Map(namespace, Map(group::serviceName, Service)).
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 thrown
service.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);
}
}
}
4.
private void putServiceAndInit(Service service) throws NacosException { putService(service); //把服務(wù)信息保存到serviceMap集合
service.init(); //建立心跳檢測機制
//實現(xiàn)數(shù)據(jù)一致性監(jiān)聽，ephemeral=true表示采用raft協(xié)議，false表示采用Distro
consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(),
service.getName(), false), service); Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson());
}
5 把服務(wù)實例添加到集合中，然后基于一致性協(xié)議進(jìn)行數(shù)據(jù)的同步
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); }
}
最終把server信息放入到servicemap里面去
java```
