通过查阅官网可知，服务注册实际上就是向Nacos服务端发起一个http请求。
对应的controller（InstanceController）如下：
@RestController
@RequestMapping(UtilsAndCommons.NACOS_NAMING_CONTEXT + UtilsAndCommons.NACOS_NAMING_INSTANCE_CONTEXT)
public class InstanceController {
。。。
}
1. 客户端服务注册流程
- nacos-discovery-spring-boot-starter 启动服务通过自动装配功能装配nacos客户端。
- Nacos自动配置服务实现Spring的应用监听器用来注册nacos服务。
- 监听到spring的WebServerInitializedEvent事件后把springboot服务注册到nacos注册中心。
- 调用nacos-client jar包中的com.alibaba.nacos.client.naming.net.NamingProxy#registerService完成服务注册。
以上为spring Boot自动装配原理以及spring容器启动时监听器的原理，不做过多解释。
registerService对应代码如下：
@Override
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", namespaceId, serviceName,
instance);
String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
if (instance.isEphemeral()) {
BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
beatReactor.addBeatInfo(groupedServiceName, beatInfo);
}
final Map<String, String> params = new HashMap<>(32);
params.put(CommonParams.NAMESPACE_ID, namespaceId);
params.put(CommonParams.SERVICE_NAME, groupedServiceName);
params.put(CommonParams.GROUP_NAME, groupName);
params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
params.put(IP_PARAM, instance.getIp());
params.put(PORT_PARAM, String.valueOf(instance.getPort()));
params.put(WEIGHT_PARAM, String.valueOf(instance.getWeight()));
params.put(REGISTER_ENABLE_PARAM, String.valueOf(instance.isEnabled()));
params.put(HEALTHY_PARAM, String.valueOf(instance.isHealthy()));
params.put(EPHEMERAL_PARAM, String.valueOf(instance.isEphemeral()));
params.put(META_PARAM, JacksonUtils.toJson(instance.getMetadata()));
reqApi(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST); //发送http请求向服务端进行注册。
}
2. 服务端注册逻辑
调用链路如下：
com.alibaba.nacos.naming.controllers.InstanceController#register
->com.alibaba.nacos.naming.core.ServiceManager#registerInstance
->com.alibaba.nacos.naming.core.ServiceManager#addInstance
->com.alibaba.nacos.naming.consistency.DelegateConsistencyServiceImpl#put
->com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl#put
->com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl#onPut
->com.alibaba.nacos.naming.consistency.ephemeral.distro.DataStore#put
->com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl.Notifier#addTask
@CanDistro
@PostMapping
@Secured(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 = HttpRequestInstanceBuilder.newBuilder()
.setDefaultInstanceEphemeral(switchDomain.isDefaultInstanceEphemeral()).setRequest(request).build();
//注册逻辑
getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
NotifyCenter.publishEvent(new RegisterInstanceTraceEvent(System.currentTimeMillis(), "",
false, namespaceId, NamingUtils.getGroupName(serviceName), NamingUtils.getServiceName(serviceName),
instance.getIp(), instance.getPort()));
return "ok";
}
接着往下走进入到InstanceOperatorServiceImpl类中
@Override
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
com.alibaba.nacos.naming.core.Instance coreInstance = parseInstance(instance);
serviceManager.registerInstance(namespaceId, serviceName, coreInstance);
}
进入到ServiceManager类中
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
NamingUtils.checkInstanceIsLegal(instance);
createEmptyService(namespaceId, serviceName, instance.isEphemeral());
Service service = getService(namespaceId, serviceName);
checkServiceIsNull(service, namespaceId, serviceName);
//添加到注册表里
addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}
在继续跟进addInstance方法
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);
}
}
依据调用链路走到这里
@Override
public void put(String key, Record value) throws NacosException {
//添加注册表
onPut(key, value);
if (ApplicationUtils.getBean(UpgradeJudgement.class).isUseGrpcFeatures()) {
return;
}
//集群架构下进行数据同步的逻辑，此分支可以先不看
distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
DistroConfig.getInstance().getSyncDelayMillis());
}
public void onPut(String key, Record value) {
if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
Datum<Instances> datum = new Datum<>();
datum.value = (Instances) value;
datum.key = key;
datum.timestamp.incrementAndGet();
//这里就是将instance保存在一个map中
dataStore.put(key, datum);
}
if (!listeners.containsKey(key)) {
return;
}
//添加客户端信息到阻塞队列
notifier.addTask(key, DataOperation.CHANGE);
}
public class Notifier implements Runnable {
private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>(10 * 1024);
//阻塞队列
private BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);
/**
* Add new notify task to queue.
*
* @param datumKey data key
* @param action action for data
*/
public void addTask(String datumKey, DataOperation action) {
if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
return;
}
if (action == DataOperation.CHANGE) {
services.put(datumKey, StringUtils.EMPTY);
}
tasks.offer(Pair.with(datumKey, action));
}
public int getTaskSize() {
return tasks.size();
}
//既然是一个线程类，那么就首先看run方法，DistroConsistencyServiceImpl初始化的时候会将Notifier
//提交到只有一个线程的线程池中去处理
@Override
public void run() {
Loggers.DISTRO.info("distro notifier started");
for (; ; ) {
try {
Pair<String, DataOperation> pair = tasks.take();
// 拿出阻塞队列中的客户端信息进行处理
handle(pair);
} catch (Throwable e) {
Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
}
}
}
private void handle(Pair<String, DataOperation> pair) {
try {
String datumKey = pair.getValue0();
DataOperation action = pair.getValue1();
services.remove(datumKey);
int count = 0;
if (!listeners.containsKey(datumKey)) {
return;
}
//遍历所有的实例
for (RecordListener listener : listeners.get(datumKey)) {
count++;
try {
//如果实例信息发生了改变
if (action == DataOperation.CHANGE) {
//在onPut方法中已经将instance放入到一个dataStore的map中，if条件满足则取出来对ip地址进行修改
listener.onChange(datumKey, dataStore.get(datumKey).value);
continue;
}
if (action == DataOperation.DELETE) {
listener.onDelete(datumKey);
continue;
}
} catch (Throwable e) {
Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);
}
}
if (Loggers.DISTRO.isDebugEnabled()) {
Loggers.DISTRO
.debug("[NACOS-DISTRO] datum change notified, key: {}, listener count: {}, action: {}",
datumKey, count, action.name());
}
} catch (Throwable e) {
Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
}
}
}
Nacos采用阻塞队列加Notifier 的形式，完成异步注册架构，这样做的好处在于：提高注册的并发，对于客户端来说就是阻塞状态，启动速度变慢，对于正常的功能没有任何影响，而且大多数项目中的服务数量也不可能存在将阻塞队列装满的情况。
后续调用链路
->com.alibaba.nacos.naming.core.Service#onChange
->com.alibaba.nacos.naming.core.Service#updateIPs
->com.alibaba.nacos.naming.core.Cluster#updateIps
public void updateIps(List<Instance> ips, boolean ephemeral) {
//如果为ephemeral 则复制出一份副本
Set<Instance> toUpdateInstances = ephemeral ? ephemeralInstances : persistentInstances;
HashMap<String, Instance> oldIpMap = new HashMap<>(toUpdateInstances.size());
//复制操作
for (Instance ip : toUpdateInstances) {
oldIpMap.put(ip.getDatumKey(), ip);
}
//基于oldIpMap 即复制出来的 进行注册操作，并不是复制出整个注册表，而是只复制了实例的set集合
List<Instance> updatedIps = updatedIps(ips, oldIpMap.values());
。。。
//最终将 toUpdateInstances 赋值给ephemeralInstances 或者 persistentInstances
toUpdateInstances = new HashSet<>(ips);
if (ephemeral) {
// Set<Instance> ephemeralInstances = new HashSet<>(); 真正存放instance的地方
ephemeralInstances = toUpdateInstances;
} else {
persistentInstances = toUpdateInstances;
}
}
此处复制出一个map的作用就是为了提高并发读写能力，利用cow的思想免除了了加锁的开销，也可以避免消费端从注册中心中读取到脏数据。又因为初始化的时候只会初始化一次，所以也只有一个线程来处理队列中的任务，所以也不会出现覆盖问题。
3. Nacos注册表结构
举例说明
