发布时间：2024-01-27 13:00
本文是个人针对nacos 2.0.3版本的解读，有错误的地方，还希望各位指正批评。
Nacos是Alibaba开源的产品，主要提供分布式微服务架构中的服务发现，配置管理，服务治理的解决方案。
Nacos主要提供了一下四大功能：
Nacos采用的是Raft算法实现集群Leader选举，和Zookeeper的选举策略大致相同，但是算法却比Zookeeper采用的选举算法更加简单明了。
Raft算法在选举的时候，有三种角色：
RaftCore为选举算法的主要实现类，入口方法在init上，有一个@PostConstruct注解，RaftCore类被加载的时候就会执行init方法。
@PostConstruct
public void init() throws Exception {
……省略部分代码……
// 注册选举任务，每500ms执行一次
masterTask = GlobalExecutor.registerMasterElection(new MasterElection());
// 注册心跳任务，每500ms执行一次
heartbeatTask = GlobalExecutor.registerHeartbeat(new HeartBeat());
……省略部分代码……
}
public void run() {
try {
// 选举任务如果已经停止就不能选举。
if (stopWork) {
return;
}
// 本机如果没有准备就绪就不能开始选举
if (!peers.isReady()) {
return;
}
// 获取本机信息
RaftPeer local = peers.local();
// leaderDueMs默认初始化为一个0~15000之间的随机数，每次选举的时候,都需要从这个数中减去500ms，之后才进行选举，
// 个人理解是为了错乱投票的时机，尽量减少同时两个节点同一时间发起投票的概率。
local.leaderDueMs -= GlobalExecutor.TICK_PERIOD_MS;
// local.leaderDueMs > 0才能进行投票
if (local.leaderDueMs > 0) {
return;
}
// 重置本机的投票时间
local.resetLeaderDue();
// 重置本机的心跳时间
local.resetHeartbeatDue();
// 发起投票
sendVote();
} catch (Exception e) {
Loggers.RAFT.warn("[RAFT] error while master election {}", e);
}
}
private void sendVote() {
// 先获取本机信息
RaftPeer local = peers.get(NetUtils.localServer());
Loggers.RAFT.info("leader timeout, start voting,leader: {}, term: {}", JacksonUtils.toJson(getLeader()),
local.term);
// 清空投票箱，leader = null, 每一个节点的投票信息值为null
peers.reset();
// 本机的任期(或者纪元)加1
local.term.incrementAndGet();
// 先给自己投一票，即本机的投票信息为自身的IP
local.voteFor = local.ip;
// 更改本机节点的状态为CANDIDATE
local.state = RaftPeer.State.CANDIDATE;
// 封装参数，准备向其他节点同步投票信息
Map<String, String> params = new HashMap<>(1);
params.put("vote", JacksonUtils.toJson(local));
// 循环向除了自身的节点之外的其他节点同步投票信息
for (final String server : peers.allServersWithoutMySelf()) {
// 构造请求URL："/raft/vote/v1/ns/raft/vote"
final String url = buildUrl(server, API_VOTE);
try {
// 发送http请求
HttpClient.asyncHttpPost(url, null, params, new Callback<String>() {
@Override
public void onReceive(RestResult<String> result) {
if (!result.ok()) {
Loggers.RAFT.error("NACOS-RAFT vote failed: {}, url: {}", result.getCode(), url);
return;
}
// 接收远程服节点返回的投票信息
RaftPeer peer = JacksonUtils.toObj(result.getData(), RaftPeer.class);
Loggers.RAFT.info("received approve from peer: {}", JacksonUtils.toJson(peer));
// 开始选举leader
peers.decideLeader(peer);
}
@Override
public void onError(Throwable throwable) {
Loggers.RAFT.error("error while sending vote to server: {}", server, throwable);
}
@Override
public void onCancel() {
}
});
} catch (Exception e) {
Loggers.RAFT.warn("error while sending vote to server: {}", server);
}
}
}
}
public RaftPeer decideLeader(RaftPeer candidate) {
// 实际上就是将每一个节点信息和对应的IP做了映射。
peers.put(candidate.ip, candidate);
// 创建了一个集合，用来所有的投票结果。
SortedBag ips = new TreeBag();
// 记录的每一个节点的投票票数
int maxApproveCount = 0;
// 记录获得最大投票票数的那个节点的IP
String maxApprovePeer = null;
for (RaftPeer peer : peers.values()) {
if (StringUtils.isEmpty(peer.voteFor)) {
continue;
}
// 将投票结果放进集合中
ips.add(peer.voteFor);
// 如果当前循环的节点的投票票数 > 本次记录的票数
if (ips.getCount(peer.voteFor) > maxApproveCount) {
// 记录的票数更改为当前循环的节点的投票票数
maxApproveCount = ips.getCount(peer.voteFor);
// 记录获得最大投票票数的那个节点的IP更改为当前循环的节点的IP
maxApprovePeer = peer.voteFor;
}
}
// 如果记录的票数 >= majorityCount(),这个方法实际上就是计算了当前集群数量的中间值，即采用的是过半机制。
if (maxApproveCount >= majorityCount()) {
// 从整个节点信息中获取票数最大的那个节点的信息。
RaftPeer peer = peers.get(maxApprovePeer);
// 将这个节点的状态更改为Leader。
peer.state = RaftPeer.State.LEADER;
// 再次处理下leader，应为已经选举出来新的leader了，这里可能就是为了处理脑裂问题吧。
if (!Objects.equals(leader, peer)) {
leader = peer;
ApplicationUtils.publishEvent(new LeaderElectFinishedEvent(this, leader, local()));
Loggers.RAFT.info("{} has become the LEADER", leader.ip);
}
}
return leader;
}
/**
* 服务端发送同步投票的信息之后，就会调用到这个方法中。
*/
public synchronized RaftPeer receivedVote(RaftPeer remote) {
// 获取当前机器的信息，这个当前机器指的是被通知投票的那个节点，而remote指的是发起投票信息的那个节点。
RaftPeer local = peers.get(NetUtils.localServer());
// 如果发起投票的那个节点的纪元比被通知投票的那个节点的纪元还小
if (remote.term.get() <= local.term.get()) {
String msg = "received illegitimate vote" + ", voter-term:" + remote.term + ", votee-term:" + local.term;
Loggers.RAFT.info(msg);
// 如果被通知投票的那个节点的投票信息是空的，那么就投给自身。
if (StringUtils.isEmpty(local.voteFor)) {
local.voteFor = local.ip;
}
// 直接返回被通知投票的那个节点，而该节点的投票信息坑定是投给自身的。
return local;
}
// 如果发起投票的那个节点的纪元比被通知投票的那个节点的纪元大
local.resetLeaderDue();
// 被通知投票的那个节点的状态更改为FOLLOWER
local.state = RaftPeer.State.FOLLOWER;
// 被通知投票的那个节点将投票投给发起投票的那个节点
local.voteFor = remote.ip;
// 被通知投票的那个节点将自身的纪元同步为发起投票的那个节点的纪元
local.term.set(remote.term.get());
Loggers.RAFT.info("vote {} as leader, term: {}", remote.ip, remote.term);
// 直接返回被通知投票的那个节点，而该节点的投票信息坑定是投给发起投票的那个节点的。
return local;
}
假设有A，B，C三个节点。
Nacos作为主流的分布式配置中心，提供了配置动态感知，动态更改的功能。
// 创建configService
public NacosConfigManager(NacosConfigProperties nacosConfigProperties) {
this.nacosConfigProperties = nacosConfigProperties;
createConfigService(nacosConfigProperties);
}
static ConfigService createConfigService(NacosConfigProperties nacosConfigProperties) {
if (Objects.isNull(service)) {
Class var1 = NacosConfigManager.class;
synchronized(NacosConfigManager.class) {
try {
if (Objects.isNull(service)) {
service = NacosFactory.createConfigService(nacosConfigProperties.assembleConfigServiceProperties());
}
} catch (NacosException var4) {
log.error(var4.getMessage());
throw new NacosConnectionFailureException(nacosConfigProperties.getServerAddr(), var4.getMessage(), var4);
}
}
}
return service;
}
public class NacosContextRefresher implements ApplicationListener<ApplicationReadyEvent>, ApplicationContextAware {
……省略部分代码……
// 实现了ApplicationListener监听类，服务启动的就会调用该方法
public void onApplicationEvent(ApplicationReadyEvent event) {
if (this.ready.compareAndSet(false, true)) {
// registerNacosListenersForApplications
this.registerNacosListenersForApplications();
}
}
private void registerNacosListenersForApplications() {
if (this.isRefreshEnabled()) {
Iterator var1 = NacosPropertySourceRepository.getAll().iterator();
while(var1.hasNext()) {
NacosPropertySource propertySource = (NacosPropertySource)var1.next();
if (propertySource.isRefreshable()) {
String dataId = propertySource.getDataId();
// 添加监听器
this.registerNacosListener(propertySource.getGroup(), dataId);
}
}
}
}
private void registerNacosListener(final String groupKey, final String dataKey) {
……省略部分代码……
try {
// 添加监听，最终会调用到ClientWorker的addTenantListeners方法中。
this.configService.addListener(dataKey, groupKey, listener);
} catch (NacosException var6) {
log.warn(String.format("register fail for nacos listener ,dataId=[%s],group=[%s]", dataKey, groupKey), var6);
}
}
……省略部分代码……
}
public ClientWorker(final ConfigFilterChainManager configFilterChainManager, ServerListManager serverListManager,
final Properties properties) throws NacosException {
this.configFilterChainManager = configFilterChainManager;
init(properties);
agent = new ConfigRpcTransportClient(properties, serverListManager);
// 长轮询线程池的定义
ScheduledExecutorService executorService = Executors
.newScheduledThreadPool(ThreadUtils.getSuitableThreadCount(1), r -> {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker");
t.setDaemon(true);
return t;
});
agent.setExecutor(executorService);
agent.start();
}
public void addTenantListeners(String dataId, String group, List<? extends Listener> listeners)
throws NacosException {
group = blank2defaultGroup(group);
String tenant = agent.getTenant();
// 根据dataId，group添加配置缓存。
CacheData cache = addCacheDataIfAbsent(dataId, group, tenant);
synchronized (cache) {
for (Listener listener : listeners) {
// 为本次配置对象添加监听器
cache.addListener(listener);
}
cache.setSyncWithServer(false);
// 实际上就是向阻塞队列中加入了一个object对象
agent.notifyListenConfig();
}
}
public CacheData addCacheDataIfAbsent(String dataId, String group, String tenant) throws NacosException {
// 先从缓存中获取，如果有直接返回
CacheData cache = getCache(dataId, group, tenant);
if (null != cache) {
return cache;
}
String key = GroupKey.getKeyTenant(dataId, group, tenant);
synchronized (cacheMap) {
CacheData cacheFromMap = getCache(dataId, group, tenant);
// 双重检查，类似于DCL。
if (null != cacheFromMap) {
cache = cacheFromMap;
cache.setInitializing(true);
} else {
// 缓存中没有，那么就构造一个CacheData对象
cache = new CacheData(configFilterChainManager, agent.getName(), dataId, group, tenant);
int taskId = cacheMap.get().size() / (int) ParamUtil.getPerTaskConfigSize();
cache.setTaskId(taskId);
// 如果需要从服务端获取缓存配置信息
if (enableRemoteSyncConfig) {
// 从服务端拉取配置信息，最终会调用到ConfigRpcTransportClient.queryConfig方法，默认发起长轮询请求获取配置，默认的Timeout时间为30s，并把返回的配置数据回填至CacheData对象的content字段，同时用content生成MD5值
ConfigResponse response = getServerConfig(dataId, group, tenant, 3000L, false);
cache.setContent(response.getContent());
}
}
// 构造好cache后，利用写时复制思想更新配置中心缓存。
Map<String, CacheData> copy = new HashMap<String, CacheData>(this.cacheMap.get());
copy.put(key, cache);
cacheMap.set(copy);
}
LOGGER.info("[{}] [subscribe] {}", agent.getName(), key);
MetricsMonitor.getListenConfigCountMonitor().set(cacheMap.get().size());
return cache;
}
public void executeConfigListen() {
Map<String, List<CacheData>> listenCachesMap = new HashMap<String, List<CacheData>>(16);
Map<String, List<CacheData>> removeListenCachesMap = new HashMap<String, List<CacheData>>(16);
long now = System.currentTimeMillis();
boolean needAllSync = now - lastAllSyncTime >= ALL_SYNC_INTERNAL;
// 服务启动的会从服务器上同步一次配置信息
for (CacheData cache : cacheMap.get().values()) {
synchronized (cache) {
//check local listeners consistent.
if (cache.isSyncWithServer()) {
cache.checkListenerMd5();
if (!needAllSync) {
continue;
}
}
// 如果配置信息中的监听器对象不为空，那么就需要将这些监听器获取出来执行
if (!CollectionUtils.isEmpty(cache.getListeners())) {
// 如果不是使用本地配置信息，那么就根据TaskId将配置信息获取出来
if (!cache.isUseLocalConfigInfo()) {
List<CacheData> cacheDatas = listenCachesMap.get(String.valueOf(cache.getTaskId()));
if (cacheDatas == null) {
cacheDatas = new LinkedList<CacheData>();
listenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
}
cacheDatas.add(cache);
}
// 如果配置信息中的监听器对象为空，那么就需要将这些监听器删除
} else if (CollectionUtils.isEmpty(cache.getListeners())) {
// 如果不是使用本地配置信息，那么就根据TaskId将配置信息获取出来
if (!cache.isUseLocalConfigInfo()) {
List<CacheData> cacheDatas = removeListenCachesMap.get(String.valueOf(cache.getTaskId()));
if (cacheDatas == null) {
cacheDatas = new LinkedList<CacheData>();
removeListenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
}
cacheDatas.add(cache);
}
}
}
}
boolean hasChangedKeys = false;
if (!listenCachesMap.isEmpty()) {
for (Map.Entry<String, List<CacheData>> entry : listenCachesMap.entrySet()) {
String taskId = entry.getKey();
Map<String, Long> timestampMap = new HashMap<>(listenCachesMap.size() * 2);
List<CacheData> listenCaches = entry.getValue();
for (CacheData cacheData : listenCaches) {
timestampMap.put(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant),
cacheData.getLastModifiedTs().longValue());
}
// 将当前taskId下的所有的配置对象信息封装成批处理请求。
ConfigBatchListenRequest configChangeListenRequest = buildConfigRequest(listenCaches);
configChangeListenRequest.setListen(true);
try {
RpcClient rpcClient = ensureRpcClient(taskId);
// 从服务端获取已经发生有变更的配置，这里获取的是发生了变成的配置信息的配置ID，并没有将变更信息获取到。
ConfigChangeBatchListenResponse configChangeBatchListenResponse = (ConfigChangeBatchListenResponse) requestProxy(
rpcClient, configChangeListenRequest);
if (configChangeBatchListenResponse != null && configChangeBatchListenResponse.isSuccess()) {
Set<String> changeKeys = new HashSet<String>();
// 如果有发生了变更了的配置ID
if (!CollectionUtils.isEmpty(configChangeBatchListenResponse.getChangedConfigs())) {
hasChangedKeys = true;
for (ConfigChangeBatchListenResponse.ConfigContext changeConfig : configChangeBatchListenResponse
.getChangedConfigs()) {
String changeKey = GroupKey
.getKeyTenant(changeConfig.getDataId(), changeConfig.getGroup(),
changeConfig.getTenant());
changeKeys.add(changeKey);
boolean isInitializing = cacheMap.get().get(changeKey).isInitializing();
// 这里会真正的获取变更信息
refreshContentAndCheck(changeKey, !isInitializing);
}
}
// 更新对应的配置信息最后被更新的时间
for (CacheData cacheData : listenCaches) {
String groupKey = GroupKey
.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.getTenant());
if (!changeKeys.contains(groupKey)) {
//sync:cache data md5 = server md5 && cache data md5 = all listeners md5.
synchronized (cacheData) {
if (!cacheData.getListeners().isEmpty()) {
Long previousTimesStamp = timestampMap.get(groupKey);
if (previousTimesStamp != null) {
if (!cacheData.getLastModifiedTs().compareAndSet(previousTimesStamp,
System.currentTimeMillis())) {
continue;
}
}
cacheData.setSyncWithServer(true);
}
}
}
cacheData.setInitializing(false);
}
}
} catch (Exception e) {
LOGGER.error("Async listen config change error ", e);
try {
Thread.sleep(50L);
} catch (InterruptedException interruptedException) {
//ignore
}
}
}
}
if (!removeListenCachesMap.isEmpty()) {
for (Map.Entry<String, List<CacheData>> entry : removeListenCachesMap.entrySet()) {
String taskId = entry.getKey();
List<CacheData> removeListenCaches = entry.getValue();
ConfigBatchListenRequest configChangeListenRequest = buildConfigRequest(removeListenCaches);
configChangeListenRequest.setListen(false);
try {
RpcClient rpcClient = ensureRpcClient(taskId);
// 服务端删除配置信息
boolean removeSuccess = unListenConfigChange(rpcClient, configChangeListenRequest);
if (removeSuccess) {
for (CacheData cacheData : removeListenCaches) {
synchronized (cacheData) {
if (cacheData.getListeners().isEmpty()) {
// 真正的删除配置信息
ClientWorker.this
.removeCache(cacheData.dataId, cacheData.group, cacheData.tenant);
}
}
}
}
} catch (Exception e) {
LOGGER.error("async remove listen config change error ", e);
}
try {
Thread.sleep(50L);
} catch (InterruptedException interruptedException) {
//ignore
}
}
}
if (needAllSync) {
lastAllSyncTime = now;
}
// 如果有变化的配置文件，那么需要重新同步客户端和服务端的配置信息
if (hasChangedKeys) {
notifyListenConfig();
}
}
public ConfigResponse queryConfig(String dataId, String group, String tenant, long readTimeouts, boolean notify)
throws NacosException {
// 构造一个request请求
ConfigQueryRequest request = ConfigQueryRequest.build(dataId, group, tenant);
request.putHeader(NOTIFY_HEADER, String.valueOf(notify));
RpcClient rpcClient = getOneRunningClient();
if (notify) {
CacheData cacheData = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
if (cacheData != null) {
// ensureRpcClient中做了三件事：1.启动rpc通信。2.启动事件消费类。3.链接服务端，并尝试和服务端同步一次数据。
rpcClient = ensureRpcClient(String.valueOf(cacheData.getTaskId()));
}
}
// 向服务端发送请求，并返回响应结果
ConfigQueryResponse response = (ConfigQueryResponse) requestProxy(rpcClient, request, readTimeouts);
ConfigResponse configResponse = new ConfigResponse();
// 如果响应成功，就开始解析响应结果。
if (response.isSuccess()) {
// 将服务端的配置信息保存在本地一份。
LocalConfigInfoProcessor.saveSnapshot(this.getName(), dataId, group, tenant, response.getContent());
configResponse.setContent(response.getContent());
String configType;
if (StringUtils.isNotBlank(response.getContentType())) {
configType = response.getContentType();
} else {
configType = ConfigType.TEXT.getType();
}
configResponse.setConfigType(configType);
String encryptedDataKey = response.getEncryptedDataKey();
// 保存加密数据密钥的快照。如果内容为NULL，会删除快照。
LocalEncryptedDataKeyProcessor
.saveEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant, encryptedDataKey);
configResponse.setEncryptedDataKey(encryptedDataKey);
return configResponse;
……省略部分代码……
}
@PostMapping("/listener")
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void listener(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
……省略部分代码……
// 长轮询逻辑
inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}
public String doPollingConfig(HttpServletRequest request, HttpServletResponse response,
Map<String, String> clientMd5Map, int probeRequestSize) throws IOException {
// 长轮询逻辑
if (LongPollingService.isSupportLongPolling(request)) {
longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
return HttpServletResponse.SC_OK + "";
}
……省略部分代码……
}
public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
int probeRequestSize) {
……省略部分代码……
// 服务端将接收到的客户端的长轮询请求封装成ClientLongPolling通过定时任务执行，每一个请求都携带一个asyncContext对象。
ConfigExecutor.executeLongPolling(
new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}
public void run() {
asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(new Runnable() {
@Override
public void run() {
……省略部分代码……
// 客户端默认30s发起一次长轮询请求，服务端线程池延后 29.5s 执行并返回响应结果，这样最大可能得避免因为客户端这边因为等待时间超过30s使得请求超时。
}, timeoutTime, TimeUnit.MILLISECONDS);
// 服务端也通过一个队列 allSubs 保存了所有正在被hold 住的轮询请求，hold住的期间，如果有配置信息发生了变化，那么就会从这个队列中找打这个ClientLongPolling请求，并且立即执行并将响应结果返回。
allSubs.add(this);
}
客户端通过NacosConfigService 类的 getConfigInner 方法请求配置信息，优先会从本地文件中查找，如果本地文件不存在，那么就从服务端获取。
客户端每一个配置信息都会被封装成一个CacheData，CacheData中有几个重要的属性content是配置内容，MD5值是用来检测配置是否发生变更的关键，内部还维护着一个若干监听器组成的数组，一旦发生变更则依次回调这些监听器，而CacheData都保存在ClientWorker类中的原子变量 cacheMap 中。
ClientWorker中配置了一个专门用于监听配置信息变化的定时线程，执行的过程中做了一下几个工作：
检查本地配置，忽略本地快照不存在的配置项，检查是否存在需要回调监听器的配置项。
如果本地没有配置项的，从服务端拿，返回配置内容发生变更的配置项ID列表。
再根据每一个配置项ID再到服务端获取最新配置，更新本地快照，补全之前缺失的配置。
检查 MD5 标签是否一致，不一致需要回调监听器。
ConfigDataChangeEvent 的事件，通知其他节点进行数据同步。LongPollingService 类中，LongPollingService 类中有一个ClientLongPolling 类，实现了Runnable接口，服务端将接收到的客户端的长轮询请求封装成ClientLongPolling通过定时任务执行，每一个请求都携带一个asyncContext对象。Nacos2.0.3版本中服务注册中心默认是AP模式，如果需要设置为CP模式，那么客户端必须设置spring.cloud.nacos.discovery.ephemeral=false （默认为true） ，表示是启用AP模式。
Nacos2.0.3版本中默认使用HTTP协议，端口号为8848 ，还增加了gRPC协议，使用gRPC协议的有两个地方，一个是客户端和服务端通信的时候，一个是集群节点之间的通信的时候。 Nacos2.0.3中gRPC的端口号是基于HTTP端口进行偏移1000生成的，即默认客户端和服务端通信的时候的端口为：8848+1000=9848，而集群之间的通信是的端口偏移1001，即8848+1001=9849，如果手动开启相关端口的话，那么这里需要开通，8848,9848,9849，三个端口号。
nacos注册中心需要增加如下依赖：
<dependency>
<groupId>com.alibaba.cloudgroupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discoveryartifactId>
dependency>
服务启动的时候，会在spring.factories中找到NacosServiceRegistryAutoConfiguration类进行注入。在该类中会自动装配NacosAutoServiceRegistration类，而NacosAutoServiceRegistration间接继承了ApplicationListener，通过初始化事件进行启动nacos服务注册功能。
public void onApplicationEvent(WebServerInitializedEvent event) {
this.bind(event);
}
// AbstractAutoServiceRegistration是NacosAutoServiceRegistration的直接父类，实现了ApplicationListener接口，重写了onApplicationEvent方法，这样在spring容器启动的时候，就会回调这个方法。this.bind(event);-》start()-》this.register();
protected void register() {
// 这个开始调用客户端注册功能准备将服务信息注册到服务端
this.serviceRegistry.register(this.getRegistration());
}
public void register(Registration registration) {
// 客户端注册服务具体实现最终要借助NamingService实现
NamingService namingService = this.namingService();
// 这里实际上就是设置的服务名称
String serviceId = registration.getServiceId();
// 属组，默认情况下为DEFAULT_GROUP
String group = this.nacosDiscoveryProperties.getGroup();
// 封装当前服务的信息，包括IP、端口等信息。
Instance instance = this.getNacosInstanceFromRegistration(registration);
// 调用注册功能
namingService.registerInstance(serviceId, group, instance);
}
}
}
// 服务的所有的信息都被封装在Instance中
private Instance getNacosInstanceFromRegistration(Registration registration) {
Instance instance = new Instance();
instance.setIp(registration.getHost());
instance.setPort(registration.getPort());
instance.setWeight((double)this.nacosDiscoveryProperties.getWeight());
instance.setClusterName(this.nacosDiscoveryProperties.getClusterName());
instance.setEnabled(this.nacosDiscoveryProperties.isInstanceEnabled());
instance.setMetadata(registration.getMetadata());
instance.setEphemeral(this.nacosDiscoveryProperties.isEphemeral());
return instance;
}
// 默认使用的是grpc协议，通过更改“spring.cloud.nacos.discovery.ephemeral=false”使用http协议
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
// 默认使用的是grpc协议，通过更改“spring.cloud.nacos.discovery.ephemeral=false”使用http协议
getExecuteClientProxy(instance).registerService(serviceName, groupName, instance);
}
// 默认使用的是grpc协议，通过更改“spring.cloud.nacos.discovery.ephemeral=false”使用http协议
private NamingClientProxy getExecuteClientProxy(Instance instance) {
return instance.isEphemeral() ? grpcClientProxy : httpClientProxy;
}
基于AP模式的时候，这里选择的getExecuteClientProxy(instance)的结果为grpcClientProxy，即使用grpc协议和服务端进行通信。
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance {}", namespaceId, serviceName,
instance);
redoService.cacheInstanceForRedo(serviceName, groupName, instance);
doRegisterService(serviceName, groupName, instance);
}
public void doRegisterService(String serviceName, String groupName, Instance instance) throws NacosException {
// 这里封装了一个InstanceRequest，后边会将这种类型的请求发送到服务端。
InstanceRequest request = new InstanceRequest(namespaceId, serviceName, groupName,
NamingRemoteConstants.REGISTER_INSTANCE, instance);
// 向服务端发送请求。
requestToServer(request, Response.class);
redoService.instanceRegistered(serviceName, groupName);
}
说明：
BeatReactor：客户端与服务端周期心跳检测类，内部定义ScheduledThreadPoolExecutor周期调度器，创建名为com.alibaba.nacos.naming.beat.sender线程周期的执行BeatTask任务(该类为BeatReactor中的一个内部类用于向服务端发送心跳信息,最终通过httpclient发送路径为/instance/beat的http请求)，内部维护以serviceName+groupName+ip+host为key，BeatInfo为value的map，当client初始化向服务端注册实例时会创建一个BeatInfo对象，通过BeatReactor中addBeatInfo()函数写入，并在beatInfo属性period(默认5s)后调度一次BeatTask。关于心跳检测可以参考后续对心跳机制描述
HostReactor：客户端周期去拉取服务端代码，内部定义ScheduledThreadPoolExecutor周期调度器，创建名为com.alibaba.nacos.client.naming.updater线程周期的执行UpdateTask任务(该类为HostReactor中的一个内部类用于更新client中缓存的服务注册列表信息，在获取列表的同时，告诉服务度它的udp端口号信息，服务端生成对应的PushClient对象，一旦服务端中对应的Service信息发生来变更，服务端可以通过PushClient进行发送变更信息。UpdateTask以Service-cluster组合为单位来周期更新的，更新频率默认1s可设置。通过updateServiceNow()发送http请求-/instance/list)
EventDispatcher：事件分发器 用于管理EventListener。内部定义ScheduledThreadPoolExecutor周期调度器，创建名为com.alibaba.nacos.naming.client.listener线程周期的执行Notifier任务(通过Notifier向注册的EventListener中发生NamingEvent事件,可用于本地扩展(实现ApplicationListener接口监控NamingEvent事件))
PushReceiver：用于接受服务端发送来的ACK数据并进行与本地信息对比更新，最后返回服务端ack信息，该类初始化时创建一个udpSocket，用于与服务端数据通信，定义ScheduledThreadPoolExecutor创建前缀名为com.alibaba.nacos.naming.push.receiver的调度器，用于执行PushReceiver(该类本身实现了Runnable接口)。
服务端grpc的实现都继承了BaseGrpcServer，而BaseGrpcServer继承了BaseRpcServer，BaseRpcServer中有一个@PostConstruct修饰的start()方法，整个rpc的启动从这里开始。
BaseGrpcServer的实现有两种实现，其中GrpcSdkServer是专门用来处理客户端和服务端之间通信的，GrpcClusterServer是专门用来处理集群间通信的。
@PostConstruct
public void start() throws Exception {
……省略部分代码……
// 启动注册中心服务
startServer();
……省略部分代码……
}
private void addServices(MutableHandlerRegistry handlerRegistry, ServerInterceptor... serverInterceptor) {
……省略部分代码……
final ServerCallHandler<Payload, Payload> payloadHandler = ServerCalls
.asyncUnaryCall((request, responseObserver) -> {
// 服务端最终使用GrpcRequestAcceptor处理客户端的request请求，GrpcRequestAcceptor会根据请求类型来确定具体的使用哪一个RequestHandler来处理，因为客户端发送请求的时候，封装的是InstanceRequest，所以在服务端使用的也就是InstanceRequestHandler来处理。
grpcCommonRequestAcceptor.request(request, responseObserver);
});
……省略部分代码……
}
public InstanceResponse handle(InstanceRequest request, RequestMeta meta) throws NacosException {
Service service = Service
.newService(request.getNamespace(), request.getGroupName(), request.getServiceName(), true);
// 根据请求类型，判断是否注册服务还是剔除服务。
switch (request.getType()) {
case NamingRemoteConstants.REGISTER_INSTANCE:
// 注册服务
return registerInstance(service, request, meta);
case NamingRemoteConstants.DE_REGISTER_INSTANCE:
return deregisterInstance(service, request, meta);
default:
throw new NacosException(NacosException.INVALID_PARAM,
String.format("Unsupported request type %s", request.getType()));
}
}
public void registerInstance(Service service, Instance instance, String clientId) {
Service singleton = ServiceManager.getInstance().getSingleton(service);
// 获取到客户端信息
Client client = clientManager.getClient(clientId);
if (!clientIsLegal(client, clientId)) {
return;
}
InstancePublishInfo instanceInfo = getPublishInfo(instance);
// AP模式下，首先会将服务信息注册到客户端
client.addServiceInstance(singleton, instanceInfo);
// 更新客户端最后同步服务信息的时间。
client.setLastUpdatedTime();
NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
NotifyCenter
.publishEvent(new MetadataEvent.InstanceMetadataEvent(singleton, instanceInfo.getMetadataId(), false));
}
// client.addServiceInstance(singleton, instanceInfo);最终会调用到AbstractClient.addServiceInstance，并发布了ClientEvent.ClientChangedEvent事件
public boolean addServiceInstance(Service service, InstancePublishInfo instancePublishInfo) {
if (null == publishers.put(service, instancePublishInfo)) {
MetricsMonitor.incrementInstanceCount();
}
// 发布ClientEvent.ClientChangedEvent事件，publishEvent方法中，会将添加到阻塞队列中的event事件拿出来不段的执行其中的onEvent方法。最终处理逻辑由DefaultPublisher实现，这是一个继承了Thread类的线程实现。
NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(this));
Loggers.SRV_LOG.info("Client change for service {}, {}", service, getClientId());
return true;
}publishEvent(final Event event) -> publishEvent(final Class<? extends Event> eventType, final Event event) -> publish(Event event)
// NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId)); -> publishEvent(final Event event) -> publishEvent(final Class eventType, final Event event) -> DefaultPublisher.publish
public boolean publish(Event event) {
checkIsStart();
// 将事件添加到阻塞队列中
boolean success = this.queue.offer(event);
if (!success) {
LOGGER.warn("Unable to plug in due to interruption, synchronize sending time, event : {}", event);
receiveEvent(event);
return true;
}
return true;
}
static {
…………省略部分代码…………
if (iterator.hasNext()) {
clazz = iterator.next().getClass();
} else {
// 静态代码块中，当NotifyCenter初始化的时候，加载了DefaultPublisher类
clazz = DefaultPublisher.class;
}
DEFAULT_PUBLISHER_FACTORY = (cls, buffer) -> {
try {
// 通过反射的方式获取DefaultPublisher类的实例。
EventPublisher publisher = clazz.newInstance();
// 执行DefaultPublisher类的init方法
publisher.init(cls, buffer);
return publisher;
} catch (Throwable ex) {
LOGGER.error("Service class newInstance has error : ", ex);
throw new NacosRuntimeException(SERVER_ERROR, ex);
}
};
…………省略部分代码…………
}
// DefaultPublisher是一个继承了Thread类的线程类。
public class DefaultPublisher extends Thread implements EventPublisher {}
// init方法中设置了线程的一些属性
public void init(Class<? extends Event> type, int bufferSize) {
// 当前线程设置为守护线程
setDaemon(true);
// 设置线程名称
setName("nacos.publisher-" + type.getName());
this.eventType = type;
this.queueMaxSize = bufferSize;
// 初始化一个阻塞队列
this.queue = new ArrayBlockingQueue<>(bufferSize);
// 调用super.start启动当前线程
start();
}
public void run() {
openEventHandler();
}
void openEventHandler() {
try {
…………省略部分代码…………
for (; ; ) {
if (shutdown) {
break;
}
// 从阻塞队列中获取event事件
final Event event = queue.take();
// 最终会执行Subscriber的onEvent方法。
receiveEvent(event);
UPDATER.compareAndSet(this, lastEventSequence, Math.max(lastEventSequence, event.sequence()));
}
} catch (Throwable ex) {
LOGGER.error("Event listener exception : ", ex);
}
}
void receiveEvent(Event event) {
for (Subscriber subscriber : subscribers) {
// 循环每一个订阅者subscriber
notifySubscriber(subscriber, event);
}
…………省略部分代码…………
}
public void notifySubscriber(final Subscriber subscriber, final Event event) {
// 将Subscriber的onEvent方法封装成一个任务。
final Runnable job = () -> subscriber.onEvent(event);
// 调用Subscriber的executor方法，这个默认实现返回的就是null。
final Executor executor = subscriber.executor();
if (executor != null) {
// 执行任务
executor.execute(job);
}
…………省略部分代码…………
}
public void onEvent(Event event) {
// 判断是否为单机启动
if (EnvUtil.getStandaloneMode()) {
return;
}
// 判断是否使用grpc通信
if (!upgradeJudgement.isUseGrpcFeatures()) {
return;
}
if (event instanceof ClientEvent.ClientVerifyFailedEvent) {
syncToVerifyFailedServer((ClientEvent.ClientVerifyFailedEvent) event);
} else {
// 将信息同步到加群中所有的节点
syncToAllServer((ClientEvent) event);
}
}
private void syncToAllServer(ClientEvent event) {
Client client = event.getClient();
…………省略部分代码…………
if (event instanceof ClientEvent.ClientDisconnectEvent) {
DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
distroProtocol.sync(distroKey, DataOperation.DELETE);
} else if (event instanceof ClientEvent.ClientChangedEvent) {
DistroKey distroKey = new DistroKey(client.getClientId(), TYPE);
// 同步服务信息的具体实现则是在DistroProtocol中，DistroProtoco实际上就是一个Distro协议的实现，最终封装的是Data changed事件任务，由DistroSyncChangeTask实现。
distroProtocol.sync(distroKey, DataOperation.CHANGE);
}
}
public void syncToTarget(DistroKey distroKey, DataOperation action, String targetServer, long delay) {
DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),
targetServer);
// 将每一个节点上的同步数据的任务封装成DistroDelayTask
DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);
// 将DistroDelayTask添加到ConcurrentHashMap集合中：ConcurrentHashMap tasks;
distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);
if (Loggers.DISTRO.isDebugEnabled()) {
Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, targetServer);
}
}
protected void processTasks() {
…………省略部分代码…………
try {
// 这里开始执行任务，默认调用的就是PushDelayTaskExecuteEngine中PushDelayTaskProcessor内部类实现
if (!processor.process(task)) {
retryFailedTask(taskKey, task);
}
} catch (Throwable e) {
getEngineLog().error("Nacos task execute error : " + e.toString(), e);
retryFailedTask(taskKey, task);
}
}
}
// processor.process(task)默认调用的就是PushDelayTaskExecuteEngine中PushDelayTaskProcessor内部类实现。
public class PushDelayTaskExecuteEngine extends NacosDelayTaskExecuteEngine {
…………省略部分代码…………
private static class PushDelayTaskProcessor implements NacosTaskProcessor {
…………省略部分代码…………
@Override
public boolean process(NacosTask task) {
PushDelayTask pushDelayTask = (PushDelayTask) task;
Service service = pushDelayTask.getService();
// 最终将task添加到TaskExecuteWorker中的任务队列中
NamingExecuteTaskDispatcher.getInstance()
.dispatchAndExecuteTask(service, new PushExecuteTask(service, executeEngine, pushDelayTask));
return true;
}
}
}
// NamingExecuteTaskDispatcher.getInstance().dispatchAndExecuteTask(service, new PushExecuteTask(service, executeEngine, pushDelayTask));最终将task添加到TaskExecuteWorker中的任务队列中
public void addTask(Object tag, AbstractExecuteTask task) {
NacosTaskProcessor processor = getProcessor(tag);
if (null != processor) {
processor.process(task);
return;
}
// 通过当前服务信息获取对应的任务
TaskExecuteWorker worker = getWorker(tag);
// 调用process将任务添加到TaskExecuteWorker中的一个阻塞队列中。
worker.process(task);
}
public boolean process(NacosTask task) {
if (task instanceof AbstractExecuteTask) {
// 调用putTask方法将任务添加到队列中。
putTask((Runnable) task);
}
return true;
}
private void putTask(Runnable task) {
try {
// 将任务添加到队列中。
queue.put(task);
} catch (InterruptedException ire) {
log.error(ire.toString(), ire);
}
}
public void run() {
while (!closed.get()) {
try {
Runnable task = queue.take();
long begin = System.currentTimeMillis();
// 默认调用的就是PushExecuteTask.run方法
task.run();
long duration = System.currentTimeMillis() - begin;
if (duration > 1000L) {
log.warn("task {} takes {}ms", task, duration);
}
} catch (Throwable e) {
log.error("[TASK-FAILED] " + e.toString(), e);
}
}
}
public void run() {
try {
PushDataWrapper wrapper = generatePushData();
for (String each : getTargetClientIds()) {
Client client = delayTaskEngine.getClientManager().getClient(each);
if (null == client) {
// means this client has disconnect
continue;
}
Subscriber subscriber = delayTaskEngine.getClientManager().getClient(each).getSubscriber(service);
// 这里就开始调用具体实现的doPushWithCallback进行集群间数据同步，默认有两种实现：PushExecutorRpcImpl和PushExecutorUdpImpl两种实现
delayTaskEngine.getPushExecutor().doPushWithCallback(each, subscriber, wrapper,
new NamingPushCallback(each, subscriber, wrapper.getOriginalData(), delayTask.isPushToAll()));
}
} catch (Exception e) {
Loggers.PUSH.error("Push task for service" + service.getGroupedServiceName() + " execute failed ", e);
delayTaskEngine.addTask(service, new PushDelayTask(service, 1000L));
}
}
public TaskExecuteWorker(final String name, final int mod, final int total, final Logger logger) {
this.name = name + "_" + mod + "%" + total;
this.queue = new ArrayBlockingQueue<Runnable>(QUEUE_CAPACITY);
this.closed = new AtomicBoolean(false);
this.log = null == logger ? LoggerFactory.getLogger(TaskExecuteWorker.class) : logger;
// 启动
new InnerWorker(name).start();
}
private class InnerWorker extends Thread {
InnerWorker(String name) {
setDaemon(false);
setName(name);
}
@Override
public void run() {
while (!closed.get()) {
try {
// 从task中获取到任务
Runnable task = queue.take();
long begin = System.currentTimeMillis();
// 执行，默认调用的就是PushExecuteTask.run方法
task.run();
long duration = System.currentTimeMillis() - begin;
if (duration > 1000L) {
log.warn("task {} takes {}ms", task, duration);
}
} catch (Throwable e) {
log.error("[TASK-FAILED] " + e.toString(), e);
}
}
}
}
private void addPublisherIndexes(Service service, String clientId) {
publisherIndexes.computeIfAbsent(service, (key) -> new ConcurrentHashSet<>());
publisherIndexes.get(service).add(clientId);
NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service, true));
}
客户端在启动的时候，将服务信息封装成instance，包含ip、端口号、服务名、集群名等信息，最终使用HttpClientRequest将请求发送出去注册服务信息，并启动一个心跳检测的定时任务，定时任务默认5s执行一次。服务端返回心跳结果中没有当前服务信息的话，则执行注册服务的逻辑。
服务端在接受到客户端的服务注册的请求之后，经过一系列的准备工作之后，因为是AP模式，只需要将服务信息放到缓存中即可。nacos的服务注册模型实际上就是一个ConcurrentHashMap：
Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
nacos在AP模式下，采用Distro协议，首先向任务阻塞对列中添加一个当前服务的实例，然后再去更新本地服务列表缓存，最后再去遍历所有的集群节点，并创建数据同步的任务，将数据同步任务添加到阻塞队列中进行数据同步，所以AP模式下，nacos不保证数据完全同步。
nacos在更新本地服务列表数据的时候，为了防止并打读写冲突，采用的写时复制技术，先将原来的服务列表集合复制一份，然后在新的服务列表集合上进行更新删除操作，最后用新的服务列表集合替换旧的服务列表集合。
nacos更新完数据之后，会发布一系列的数据更改事件，通知客户端服务数据已经发生改变，并发送UDP数据包到客户端，客户端收到UDP数据后会反馈一个ACK给服务端，如果服务端没有收到ACK，那么再次尝试发送UDP数据包到客户端，如果重试时间超出规定时间依旧没有收到ACK，那么就不在发送了。
客户端也会通过定时任务不断的从服务端拉取服务列表信息来更新自己的本地缓存。
客户端也会通过心跳机制上报自己的健康信息，默认是5s上报一次，如果nacos服务端15秒内未收到客户端的心跳，那么就将该服务标记为不健康状态，之后超过30秒还未收到心跳上报，那么就会删除这个服务，如果后续这个服务又重新启动，那么就需要走注册流程。
nacos服务端也会主动进行心跳检测，默认时间间隔为20s。
nacos和client之间采取推拉结合的交互方式，一方面client可以通过定时任务每隔10s向nacos发起查询请求，如果服务列表改变nacos就会返回新列表，另一方面当本地服务实例发生变化时（即server实例注册成功或者心跳停止断开链接），nacos会主动通过UDP协议推送到client，udp协议非常快，不需要保持长连接。客户端拉取和服务器推送是互补的，能在一定程度上保证数据的时效性并提高性能。
基于AP模式的时候，这里选择的getExecuteClientProxy(instance)的结果为httpClientProxy，即使用http协议和服务端进行通信。
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", namespaceId, serviceName,
instance);
// 将服务名称和分组名进行拼接：DEFAULT_GROUP@@cloud-reading-accoud-server
String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
// 如果是临时节点，需要创建心跳检测任务。心跳任务使用一个定时任务来处理，默认5s执行一次。服务端返回心跳结果总没有目前的一个服务信息，则执行注册服务的逻辑。
if (instance.isEphemeral()) {
BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
beatReactor.addBeatInfo(groupedServiceName, beatInfo);
}
// 请求参数配置
final Map<String, String> params = new HashMap<String, String>(16);
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("enable", 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()));
// UtilAndComs.nacosUrlInstance就是拼装的后端的地址，使用POST请求方式处理
// public static String webContext = "/nacos";
// public static String nacosUrlBase = webContext + "/v1/ns";
// public static String nacosUrlInstance = nacosUrlBase + "/instance";
reqApi(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST);
}
// reqApi最终会调用callServer向服务端发送请求。
public String callServer(String api, Map<String, String> params, Map<String, String> body, String curServer,
String method) throws NacosException {
try {
// 实际上就是利用HttpClient发送请求。
HttpRestResult<String> restResult = nacosRestTemplate
.exchangeForm(url, header, Query.newInstance().initParams(params), body, method, String.class);
throw new NacosException(restResult.getCode(), restResult.getMessage());
} catch (Exception e) {
NAMING_LOGGER.error("[NA] failed to request", e);
throw new NacosException(NacosException.SERVER_ERROR, e);
}
}
public void run() {
while (!closed) {
try {
// byte[] is initialized with 0 full filled by default
byte[] buffer = new byte[UDP_MSS];
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
udpSocket.receive(packet);
String json = new String(IoUtils.tryDecompress(packet.getData()), UTF_8).trim();
NAMING_LOGGER.info("received push data: " + json + " from " + packet.getAddress().toString());
PushPacket pushPacket = JacksonUtils.toObj(json, PushPacket.class);
String ack;
if (PUSH_PACKAGE_TYPE_DOM.equals(pushPacket.type) || PUSH_PACKAGE_TYPE_SERVICE.equals(pushPacket.type)) {
// 这里会将服务器服务注册信息同步到客户端
serviceInfoHolder.processServiceInfo(pushPacket.data);
// send ack to server
ack = "{\"type\": \"push-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":"
+ "\"\"}";
} else if (PUSH_PACKAGE_TYPE_DUMP.equals(pushPacket.type)) {
// dump data to server
ack = "{\"type\": \"dump-ack\"" + ", \"lastRefTime\": \"" + pushPacket.lastRefTime + "\", \"data\":"
+ "\"" + StringUtils.escapeJavaScript(JacksonUtils.toJson(serviceInfoHolder.getServiceInfoMap()))
+ "\"}";
} else {
// do nothing send ack only
ack = "{\"type\": \"unknown-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
+ "\", \"data\":" + "\"\"}";
}
// 向服务端返回ack信息
udpSocket.send(new DatagramPacket(ack.getBytes(UTF_8), ack.getBytes(UTF_8).length,
packet.getSocketAddress()));
} catch (Exception e) {
if (closed) {
return;
}
NAMING_LOGGER.error("[NA] error while receiving push data", e);
}
}
}
public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
………省略部分代码………
// 同步服务信息
serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
………省略部分代码………
}
public String register(HttpServletRequest request) throws Exception {
// 这里还是获取命名空间，默认就是public
final String namespaceId = WebUtils
.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
// 获取客户端拼接的服务名称：DEFAULT_GROUP@@cloud-reading-accoud-server
final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
NamingUtils.checkServiceNameFormat(serviceName);
// 从请求中将参数封装成instance实例。
final Instance instance = HttpRequestInstanceBuilder.newBuilder() .setDefaultInstanceEphemeral(switchDomain.isDefaultInstanceEphemeral()).setRequest(request).build();
// getInstanceOperator()：获取instance服务实例去处理具体的注册事宜，默认如果是单机版的nacos服务，那么就使用InstanceOperatorClientImpl处理，否则就使用InstanceOperatorServiceImpl。
getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
return "ok";
}
public void registerInstance(String namespaceId, String serviceName, Instance instance) {
// 这里因为在客户端设置了false，走的http协议
boolean ephemeral = instance.isEphemeral();
// 拼装了下客户端ID：10.40.177.238:9008#false
String clientId = IpPortBasedClient.getClientId(instance.toInetAddr(), ephemeral);
createIpPortClientIfAbsent(clientId);
// 获取的service实际上包含了：服务名称，属组，版本等信息，但是此时只是一个空的服务，还没有具体的shili。
Service service = getService(namespaceId, serviceName, ephemeral);
// 给服务添加具体的实例。
clientOperationService.registerInstance(service, instance, clientId);
}
public void registerInstance(Service service, Instance instance, String clientId) {
final ClientOperationService operationService = chooseClientOperationService(instance);
operationService.registerInstance(service, instance, clientId);
}
public void registerInstance(Service service, Instance instance, String clientId) {
final InstanceStoreRequest request = new InstanceStoreRequest();
request.setService(service);
request.setInstance(instance);
request.setClientId(clientId);
final WriteRequest writeRequest = WriteRequest.newBuilder().setGroup(group())
.setData(ByteString.copyFrom(serializer.serialize(request))).setOperation(DataOperation.ADD.name())
.build();
try {
// 实际调用的是JRaftProtocol.write
protocol.write(writeRequest);
} catch (Exception e) {
throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);
}
}
public Response write(WriteRequest request) throws Exception {
// 会调用JRaftServer.commit进行信息同步
CompletableFuture<Response> future = writeAsync(request);
return future.get(10_000L, TimeUnit.MILLISECONDS);
}
public CompletableFuture<Response> commit(final String group, final Message data,
final CompletableFuture<Response> future) {
………省略部分代码………
final Node node = tuple.node;
if (node.isLeader()) {
// 如果当前节点是leader，那么直接写入数据
applyOperation(node, data, closure);
} else {
// 如果当前节点不是leader，那么就寻找leader，然后将请求转发到leader上，最终进行数据写入。
invokeToLeader(group, data, rpcRequestTimeoutMs, closure);
}
return future;
}
public void applyOperation(Node node, Message data, FailoverClosure closure) {
final Task task = new Task();
task.setDone(new NacosClosure(data, status -> {
NacosClosure.NacosStatus nacosStatus = (NacosClosure.NacosStatus) status;
closure.setThrowable(nacosStatus.getThrowable());
closure.setResponse(nacosStatus.getResponse());
closure.run(nacosStatus);
}));
task.setData(ByteBuffer.wrap(data.toByteArray()));
// 最终调用的是NodeImpl.apply，封装任务，执行写入数据，
node.apply(task);
}
NacosStateMachine.onApply方法。public void onApply(Iterator iter) {
try {
while (iter.hasNext()) {
Status status = Status.OK();
try {
………省略部分代码………
if (message instanceof WriteRequest) {
// 最终会调用PersistentClientOperationServiceImpl.onApply方法：
Response response = processor.onApply((WriteRequest) message);
postProcessor(response, closure);
}
………省略部分代码………
} catch (Throwable e) {
} finally {
Optional.ofNullable(closure).ifPresent(closure1 -> closure1.run(status));
}
………省略部分代码………
}
public Response onApply(WriteRequest request) {
final InstanceStoreRequest instanceRequest = serializer.deserialize(request.getData().toByteArray());
final DataOperation operation = DataOperation.valueOf(request.getOperation());
final Lock lock = readLock;
lock.lock();
try {
switch (operation) {
// 注册服务，走ADD分支
case ADD:
onInstanceRegister(instanceRequest.service, instanceRequest.instance,
instanceRequest.getClientId());
break;
case DELETE:
onInstanceDeregister(instanceRequest.service, instanceRequest.getClientId());
break;
default:
return Response.newBuilder().setSuccess(false).setErrMsg("unsupport operation : " + operation)
.build();
}
return Response.newBuilder().setSuccess(true).build();
} finally {
lock.unlock();
}
}
private void onInstanceRegister(Service service, Instance instance, String clientId) {
Service singleton = ServiceManager.getInstance().getSingleton(service);
if (!clientManager.contains(clientId)) {
clientManager.clientConnected(clientId, new ClientAttributes());
}
Client client = clientManager.getClient(clientId);
InstancePublishInfo instancePublishInfo = getPublishInfo(instance);
// 这里在 client.addServiceInstance逻辑中会发布相关事件，通知服务消费端服务的变更信息，和AP模式一样也是通过PushExecutorDelegate代理来选择通过什么方式推送给消费客户端，而选择的逻辑则是判断建立连接的clientId是否包含"#"，如果包含则采用UDP的实现方式，如果不包含则使用rpc的实现方式，cp模式下，默认使用rpc的实现进行推送，也就是RpcPushService.pushWithCallback具体实现。客户端在启动时候会开启PushReceiver线程进行监控服务端同步服务注册信息的命令。
client.addServiceInstance(singleton, instancePublishInfo);
client.setLastUpdatedTime();
NotifyCenter.publishEvent(new ClientOperationEvent.ClientRegisterServiceEvent(singleton, clientId));
}
客户端在启动的时候，将服务信息封装成instance，包含ip、端口号、服务名、集群名等信息，最终使用HttpClientRequest将请求发送出去注册服务信息。
服务端在接受到客户端的服务注册的请求之后，经过一系列的准备工作之后，因为是CP模式，leader接收到服务信息后，立即写入。nacos的服务注册模型实际上就是一个ConcurrentHashMap：
Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();
nacos在更新本地服务列表数据的时候，为了防止并打读写冲突，采用的写时复制技术，先将原来的服务列表集合复制一份，然后在新的服务列表集合上进行更新删除操作，最后用新的服务列表集合替换旧的服务列表集合。
nacos更新完数据之后，会发布一系列的数据更改事件，通知客户端服务数据已经发生改变，并发送UDP数据包到客户端，客户端收到UDP数据后会反馈一个ACK给服务端，如果服务端没有收到ACK，那么再次尝试发送UDP数据包到客户端，如果重试时间超出规定时间依旧没有收到ACK，那么就不在发送了。
客户端也会通过定时任务不断的从服务端拉取服务列表信息来更新自己的本地缓存。
nacos服务端也会主动进行心跳检测，默认时间间隔为20s。
在nacos2.0.3版本下，服务注册功能在服务端使用AP模式还是使用CP模式，取决于客户端的配置（spring.cloud.nacos.discovery.ephemeral），默认为AP模式。
对于服务端来说，再启动的时候，同时启动了HTTP和GRPC两种模式的通信方式，客户端和服务端的具体使用哪种协议交互取取决于客户端参数控制，不管是CP模式还是AP模式，服务端在处理服务注册信息变更时，逻辑基本上都是一样的，只不过，AP模式下，写服务信息时采用了Raft协议，必须等到过半数的节点写数据成功后leader才会真正写入成功，而CP模式就是一上来就直接在leader上写入服务信息，所以CP模式不能保证所有节点的信息完全同步。另外不管CP还是AP模式获取服务注册成功都会向ClientServiceIndexesManager和ClientManager写入信息，需要获取某个服务的所有的实例的时候，通过ClientServiceIndexesManager和Service能够获取到所有的该服务实例的clientId，然后通过ClientManager和clientId集合能够获取到该服务所有实例节点信息。
客户端在启动时候会开启PushReceiver线程进行监控服务端同步服务注册信息的命令，也就是UDP监听方式。
Instance：nacos数据模型中最小存储单位，主要由ip+port来确定唯一性，一个client对应一个instance
Cluster：集群由相同的配置的Instance构成，内部存储了该集群下persistentInstances与ephemeralInstances。Cluster实例完成之后通过HealthCheckTask来检测所属Instance活性
Service：代表一个服务实例，由多个Cluster构，实现了RecordListener接口，当服务下的实例集群发生变化(增删该，通过事件驱动来解耦)，触发其onChange事件，更新本服务器实例信息之后，同时通过PushService来推送注册到其实例下的client变更的节点信息
namespace：命名空间，可以使用它来管理一个注册中心管理多个环境（开发，测试(不同环境)，生产）等
DistroConsistencyServiceImpl：AP模型体现，所有节点都是对等的(nacos这里是参考了Eureka服务的原理)，负责管理ephemeral实例信息(数据存储在内存中，由DataStore存储，初始化时会执行一次LoadDataTask任务来同步其它服务上的已存储的ephemeral实例信息，通过定期执行同步任务LoadDataTask，与其它节点做到最终数据统一，一旦实例数据发生变化触发通知机制，不仅通知其它节点也会push发送客户端，也通过TaskDispatcher来添加任务最终与其它节点进行同步) .内部包含一个TaskDispatcher来管理一批TaskScheduler的执行，而TaskScheduler通过BlockingQueue来存储数据变更同步到其它服务节点任务。最终使得集群中临时节点实例信息达到最终一致。
RaftConsistencyServiceImpl：CP模型体现，这里nacos对Raft算法的实现。除了leader写入后，follower过半数统一后再写入，follower也会通过leader心跳定期同步数据。负责保存persistent实例信息(数据存储在本地文件磁盘中，由RaftStore存储，实际上在RaftCore中会存在一份缓存数据用于读，避免频繁读写磁盘，写的动作在写入磁盘时刷新缓存)
nacos中心跳检测一般指客户端在注册临时节点的会启动一个定时线程，这个定时线程会不断的向服务端发送心跳请求以达到和服务端保持通信。
// 如果是临时节点，那么就需要创建心跳检查任务。
if (instance.isEphemeral()) {
// 构造心跳模型，包括实例IP，实例端口等一些信息
BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
// 开启心跳任务
beatReactor.addBeatInfo(groupedServiceName, beatInfo);
}
public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
// 这里实际上就是拼接了一个字符串
// serviceName + # + ip + # + port
String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
BeatInfo existBeat;
//fix #1733
if ((existBeat = dom2Beat.remove(key)) != null) {
existBeat.setStopped(true);
}
dom2Beat.put(key, beatInfo);
// 创建一个BeatTask任务，并将任务交给线程池执行，默认延迟5s执行一次
executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
}
public void run() {
if (beatInfo.isStopped()) {
return;
}
// 获取设置的延迟时间
long nextTime = beatInfo.getPeriod();
try {
// 调用NamingHttpClientProxy.sendBeat，发送心跳请求
JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
long interval = result.get(CLIENT_BEAT_INTERVAL_FIELD).asLong();
boolean lightBeatEnabled = false;
if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) {
lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean();
}
BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
if (interval > 0) {
// 设置下一次发送心跳请求的时间
nextTime = interval;
}
int code = NamingResponseCode.OK;
if (result.has(CommonParams.CODE)) {
code = result.get(CommonParams.CODE).asInt();
}
// 如果服务端返回结果是无此实例，那么就构造当前实例信息
if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
Instance instance = new Instance();
instance.setPort(beatInfo.getPort());
instance.setIp(beatInfo.getIp());
instance.setWeight(beatInfo.getWeight());
instance.setMetadata(beatInfo.getMetadata());
instance.setClusterName(beatInfo.getCluster());
instance.setServiceName(beatInfo.getServiceName());
instance.setInstanceId(instance.getInstanceId());
instance.setEphemeral(true);
try {
// 调用服务注册功能将此实例注册上去
serverProxy.registerService(beatInfo.getServiceName(),
NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
} catch (Exception ignore) {
}
}
} catch (NacosException ex) {
NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",
JacksonUtils.toJson(beatInfo), ex.getErrCode(), ex.getErrMsg());
} catch (Exception unknownEx) {
NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, unknown exception msg: {}",
JacksonUtils.toJson(beatInfo), unknownEx.getMessage(), unknownEx);
} finally {
// 最终继续封装BeatTask任务交给线程池处理。
executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
}
}
// 发送心跳请求
public JsonNode sendBeat(BeatInfo beatInfo, boolean lightBeatEnabled) throws NacosException {
if (NAMING_LOGGER.isDebugEnabled()) {
NAMING_LOGGER.debug("[BEAT] {} sending beat to server: {}", namespaceId, beatInfo.toString());
}
Map<String, String> params = new HashMap<String, String>(8);
Map<String, String> bodyMap = new HashMap<String, String>(2);
if (!lightBeatEnabled) {
bodyMap.put("beat", JacksonUtils.toJson(beatInfo));
}
params.put(CommonParams.NAMESPACE_ID, namespaceId);
params.put(CommonParams.SERVICE_NAME, beatInfo.getServiceName());
params.put(CommonParams.CLUSTER_NAME, beatInfo.getCluster());
params.put(IP_PARAM, beatInfo.getIp());
params.put(PORT_PARAM, String.valueOf(beatInfo.getPort()));
// 调用reqApi发送请求
String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/beat", params, bodyMap, HttpMethod.PUT);
return JacksonUtils.toObj(result);
}
/**
* Create a beat for instance.
*
* @param request http request
* @return detail information of instance
* @throws Exception any error during handle
*/
@CanDistro
@PutMapping("/beat")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public ObjectNode beat(HttpServletRequest request) throws Exception {
ObjectNode result = JacksonUtils.createEmptyJsonNode();
result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());
// 获取beat心跳请求标识
String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
RsInfo clientBeat = null;
if (StringUtils.isNotBlank(beat)) {
clientBeat = JacksonUtils.toObj(beat, RsInfo.class);
}
// 获取集群名称
String clusterName = WebUtils
.optional(request, CommonParams.CLUSTER_NAME, UtilsAndCommons.DEFAULT_CLUSTER_NAME);
// 获取IP
String ip = WebUtils.optional(request, "ip", StringUtils.EMPTY);
// 获取端口
int port = Integer.parseInt(WebUtils.optional(request, "port", "0"));
if (clientBeat != null) {
if (StringUtils.isNotBlank(clientBeat.getCluster())) {
clusterName = clientBeat.getCluster();
} else {
// fix #2533
clientBeat.setCluster(clusterName);
}
ip = clientBeat.getIp();
port = clientBeat.getPort();
}
// 空间名称
String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
// 服务名称
String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
NamingUtils.checkServiceNameFormat(serviceName);
Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}, namespaceId: {}", clientBeat,
serviceName, namespaceId);
BeatInfoInstanceBuilder builder = BeatInfoInstanceBuilder.newBuilder();
builder.setRequest(request);
int resultCode = getInstanceOperator()
// 调用InstanceOperatorClientImpl.handleBeat处理心跳请求
.handleBeat(namespaceId, serviceName, ip, port, clusterName, clientBeat, builder);
result.put(CommonParams.CODE, resultCode);
result.put(SwitchEntry.CLIENT_BEAT_INTERVAL,
getInstanceOperator().getHeartBeatInterval(namespaceId, serviceName, ip, port, clusterName));
result.put(SwitchEntry.LIGHT_BEAT_ENABLED, switchDomain.isLightBeatEnabled());
return result;
}
public int handleBeat(String namespaceId, String serviceName, String ip, int port, String cluster,
RsInfo clientBeat, BeatInfoInstanceBuilder builder) throws NacosException {
Service service = getService(namespaceId, serviceName, true);
String clientId = IpPortBasedClient.getClientId(ip + InternetAddressUtil.IP_PORT_SPLITER + port, true);
IpPortBasedClient client = (IpPortBasedClient) clientManager.getClient(clientId);
if (null == client || !client.getAllPublishedService().contains(service)) {
if (null == clientBeat) {
return NamingResponseCode.RESOURCE_NOT_FOUND;
}
Instance instance = builder.setBeatInfo(clientBeat).setServiceName(serviceName).build();
// 心跳检测的时候，如果根据服务信息，找不到对应的客户端信息，那就说明，客户端未注册，也会调用注册服务的逻辑
registerInstance(namespaceId, serviceName, instance);
client = (IpPortBasedClient) clientManager.getClient(clientId);
}
if (!ServiceManager.getInstance().containSingleton(service)) {
throw new NacosException(NacosException.SERVER_ERROR,
"service not found: " + serviceName + "@" + namespaceId);
}
if (null == clientBeat) {
clientBeat = new RsInfo();
clientBeat.setIp(ip);
clientBeat.setPort(port);
clientBeat.setCluster(cluster);
clientBeat.setServiceName(serviceName);
}
// 心跳检查实现
ClientBeatProcessorV2 beatProcessor = new ClientBeatProcessorV2(namespaceId, clientBeat, client);
HealthCheckReactor.scheduleNow(beatProcessor);
client.setLastUpdatedTime();
return NamingResponseCode.OK;
}
public void run() {
if (Loggers.EVT_LOG.isDebugEnabled()) {
Loggers.EVT_LOG.debug("[CLIENT-BEAT] processing beat: {}", rsInfo.toString());
}
String ip = rsInfo.getIp();
int port = rsInfo.getPort();
String serviceName = NamingUtils.getServiceName(rsInfo.getServiceName());
String groupName = NamingUtils.getGroupName(rsInfo.getServiceName());
Service service = Service.newService(namespace, groupName, serviceName, rsInfo.isEphemeral());
// 首先根据服务信息，从服务端获取当前实例信息。
HealthCheckInstancePublishInfo instance = (HealthCheckInstancePublishInfo) client.getInstancePublishInfo(service);
// 如果当前实例的IP和端口一致，说明这个实例存在
if (instance.getIp().equals(ip) && instance.getPort() == port) {
if (Loggers.EVT_LOG.isDebugEnabled()) {
Loggers.EVT_LOG.debug("[CLIENT-BEAT] refresh beat: {}", rsInfo);
}
// 更新实例和服务端最后心跳时间。
instance.setLastHeartBeatTime(System.currentTimeMillis());
// 如果当前健康状态为不健康，之前出现过故障
if (!instance.isHealthy()) {
// 将当前实例健康状态更新为健康
instance.setHealthy(true);
Loggers.EVT_LOG.info("service: {} {POS} {IP-ENABLED} valid: {}:{}@{}, region: {}, msg: client beat ok",
rsInfo.getServiceName(), ip, port, rsInfo.getCluster(), UtilsAndCommons.LOCALHOST_SITE);
// 发布服务变更事件
NotifyCenter.publishEvent(new ServiceEvent.ServiceChangedEvent(service));
NotifyCenter.publishEvent(new ClientEvent.ClientChangedEvent(client));
}
}
}
服务端对于每一个Sevice都会创建一个健康检查的延时任务，默认是刚启动后延迟5s执行，之后每5s执行一次。
public void init() {
// 服务端健康检查入口，clientBeatCheckTask是一个线程。
HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
entry.getValue().setService(this);
entry.getValue().init();
}
}
public void run() {
// 获取当前服务的所有的实例
List<Instance> instances = service.allIPs(true);
// 循环遍历所有的实例
for (Instance instance : instances) {
// 如果距离该实例的最后心跳时间超过15s
if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {
if (!instance.isMarked()) {
if (instance.isHealthy()) {
// 将该实例的健康状态改为false
instance.setHealthy(false);
// 发布服务更改事件通知
getPushService().serviceChanged(service);
}
}
}
}
// 再次循环检查各个实例
for (Instance instance : instances) {
if (instance.isMarked()) {
continue;
}
// 如果距离该实例的最后心跳时间超过30s
if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {
// 就需要删除该实例，实际就是通过http调用删除实例的接口
deleteIp(instance);
}
}
}