1、获取服务信息
HostReactor 中的 getServiceInfo( ) 方法
public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
String key = ServiceInfo.getKey(serviceName, clusters);
//开启故障转移模式，从failoverReactor 获取服务列表
if (failoverReactor.isFailoverSwitch()) {
return failoverReactor.getService(key);
}
//从缓存中获取服务列表
ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
if (null == serviceObj) {
//服务信息不存在时，创建新的服务信息，并放入缓存serviceInfoMap
serviceObj = new ServiceInfo(serviceName, clusters);
serviceInfoMap.put(serviceObj.getKey(), serviceObj);
//记录服务正在更新中
updatingMap.put(serviceName, new Object());
//更新服务
updateServiceNow(serviceName, clusters);
//服务更新完删除缓存
updatingMap.remove(serviceName);
} else if (updatingMap.containsKey(serviceName)) {
//服务正在更新中，wait 5s UPDATE_HOLD_INTERVAL = 5000L
if (UPDATE_HOLD_INTERVAL > 0) {
// hold a moment waiting for update finish
synchronized (serviceObj) {
try {
serviceObj.wait(UPDATE_HOLD_INTERVAL);
} catch (InterruptedException e) {
NAMING_LOGGER
.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
}
}
}
}
//服务更新任务不存在时，创建服务更新任务并调度
scheduleUpdateIfAbsent(serviceName, clusters);
return serviceInfoMap.get(serviceObj.getKey());
}
2、故障转移
//failover-mode参数是由SwitchRefresher来处理的
public boolean isFailoverSwitch() {
return Boolean.parseBoolean(switchParams.get("failover-mode"));
}
//serviceMap是由FailoverFileReader来处理的
public ServiceInfo getService(String key) {
ServiceInfo serviceInfo = serviceMap.get(key);
if (serviceInfo == null) {
serviceInfo = new ServiceInfo();
serviceInfo.setName(key);
}
return serviceInfo;
}
3、从缓存中获取服务信息
private ServiceInfo getServiceInfo0(String serviceName, String clusters) {
String key = ServiceInfo.getKey(serviceName, clusters);
//从 serviceInfoMap 中获取服务信息
return serviceInfoMap.get(key);
}
4、从远程更新服务信息
当缓存中不存在服务信息时，更新服务
1、updateServiceNow( )
private void updateServiceNow(String serviceName, String clusters) {
try {
updateService(serviceName, clusters);
} catch (NacosException e) {
NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
}
}
2、updateService( )
public void updateService(String serviceName, String clusters) throws NacosException {
//获取远程服务列表前，再次查询缓存中是否存在服务信息，如果存在，在finally里将等待服务列表的线程唤醒，在下面第5步中讲解等待的线程
ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
try {
String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);
if (StringUtils.isNotEmpty(result)) {
processServiceJson(result);
}
} finally {
if (oldService != null) {
synchronized (oldService) {
oldService.notifyAll();
}
}
}
}
3、queryList( )
调用nacos服务端远程接口获取对应的服务信息
public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly)
throws NacosException {
final Map<String, String> params = new HashMap<String, String>(8);
params.put(CommonParams.NAMESPACE_ID, namespaceId);
params.put(CommonParams.SERVICE_NAME, serviceName);
params.put("clusters", clusters);
params.put("udpPort", String.valueOf(udpPort));
params.put("clientIP", NetUtils.localIP());
params.put("healthyOnly", String.valueOf(healthyOnly));
return reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);
}
4、processServiceJson( )
处理上一步的返回服务信息
public ServiceInfo processServiceJson(String json) {
ServiceInfo serviceInfo = JacksonUtils.toObj(json, ServiceInfo.class);
ServiceInfo oldService = serviceInfoMap.get(serviceInfo.getKey());
//空或者返回的服务信息异常时，返回旧的服务信息
if (pushEmptyProtection && !serviceInfo.validate()) {
//empty or error push, just ignore
return oldService;
}
boolean changed = false;
if (oldService != null) {
if (oldService.getLastRefTime() > serviceInfo.getLastRefTime()) {
NAMING_LOGGER.warn("out of date data received, old-t: " + oldService.getLastRefTime() + ", new-t: "
+ serviceInfo.getLastRefTime());
}
//将新的服务信息放入缓存
serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
//旧的服务主机Map
Map<String, Instance> oldHostMap = new HashMap<String, Instance>(oldService.getHosts().size());
for (Instance host : oldService.getHosts()) {
oldHostMap.put(host.toInetAddr(), host);
}
//新的服务主机Map
Map<String, Instance> newHostMap = new HashMap<String, Instance>(serviceInfo.getHosts().size());
for (Instance host : serviceInfo.getHosts()) {
newHostMap.put(host.toInetAddr(), host);
}
//修改了的主机
Set<Instance> modHosts = new HashSet<Instance>();
//新上线的主机
Set<Instance> newHosts = new HashSet<Instance>();
//下线的主机
Set<Instance> remvHosts = new HashSet<Instance>();
List<Map.Entry<String, Instance>> newServiceHosts = new ArrayList<Map.Entry<String, Instance>>(
newHostMap.entrySet());
for (Map.Entry<String, Instance> entry : newServiceHosts) {
Instance host = entry.getValue();
String key = entry.getKey();
if (oldHostMap.containsKey(key) && !StringUtils
.equals(host.toString(), oldHostMap.get(key).toString())) {
modHosts.add(host);
continue;
}
if (!oldHostMap.containsKey(key)) {
newHosts.add(host);
}
}
for (Map.Entry<String, Instance> entry : oldHostMap.entrySet()) {
Instance host = entry.getValue();
String key = entry.getKey();
if (newHostMap.containsKey(key)) {
continue;
}
if (!newHostMap.containsKey(key)) {
remvHosts.add(host);
}
}
//记录服务主机变化日志信息
if (newHosts.size() > 0) {
changed = true;
NAMING_LOGGER.info("new ips(" + newHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
+ JacksonUtils.toJson(newHosts));
}
if (remvHosts.size() > 0) {
changed = true;
NAMING_LOGGER.info("removed ips(" + remvHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
+ JacksonUtils.toJson(remvHosts));
}
if (modHosts.size() > 0) {
changed = true;
updateBeatInfo(modHosts);
NAMING_LOGGER.info("modified ips(" + modHosts.size() + ") service: " + serviceInfo.getKey() + " -> "
+ JacksonUtils.toJson(modHosts));
}
serviceInfo.setJsonFromServer(json);
if (newHosts.size() > 0 || remvHosts.size() > 0 || modHosts.size() > 0) {
//发布服务变化事件 InstancesChangeEvent
NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(),
serviceInfo.getClusters(), serviceInfo.getHosts()));
//写入缓村文件，如果开启了loadCacheAtStart，会在系统启动时从缓村加载服务信息
DiskCache.write(serviceInfo, cacheDir);
}
} else {
//该分支是旧的服务不存在时
changed = true;
NAMING_LOGGER.info("init new ips(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> "
+ JacksonUtils.toJson(serviceInfo.getHosts()));
//服务信息存入内存缓存
serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
//发布InstancesChangeEvent事件
NotifyCenter.publishEvent(new InstancesChangeEvent(serviceInfo.getName(), serviceInfo.getGroupName(),
serviceInfo.getClusters(), serviceInfo.getHosts()));
serviceInfo.setJsonFromServer(json);
//写入缓存文件
DiskCache.write(serviceInfo, cacheDir);
}
//prometheus 监控统计
MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
if (changed) {
NAMING_LOGGER.info("current ips:(" + serviceInfo.ipCount() + ") service: " + serviceInfo.getKey() + " -> "
+ JacksonUtils.toJson(serviceInfo.getHosts()));
}
return serviceInfo;
}
5、服务正在更新中
等待5s时间，在等待过程中，如果服务更新完成，会发出通知唤醒等待的线程
if (UPDATE_HOLD_INTERVAL > 0) {
// hold a moment waiting for update finish
synchronized (serviceObj) {
try {
serviceObj.wait(UPDATE_HOLD_INTERVAL);
} catch (InterruptedException e) {
NAMING_LOGGER
.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
}
}
}
6、定时更新服务
1、scheduleUpdateIfAbsent( )
public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
//缓存中服务是否存在
if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
return;
}
synchronized (futureMap) {
//双重检验缓存
if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
return;
}
//服务更新任务 UpdateTask
ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
}
}
public synchronized ScheduledFuture<?> addTask(UpdateTask task) {
//延时1s调度任务 DEFAULT_DELAY = 1000L
return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS);
}
2、UpdateTask
public void run() {
long delayTime = DEFAULT_DELAY;
try {
ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
if (serviceObj == null) {
//服务信息不存在，更新服务信息
updateService(serviceName, clusters);
return;
}
if (serviceObj.getLastRefTime() <= lastRefTime) {
//服务信息更新时间不变，即跟上次相比服务没有更新
//更新服务信息
updateService(serviceName, clusters);
serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
} else {
// if serviceName already updated by push, we should not override it
// since the push data may be different from pull through force push
//距离上次时间，服务已经更新过，只进行服务刷新，不处理拉取下来的新的服务
refreshOnly(serviceName, clusters);
}
//更新上次服务信息返回的时间字段
lastRefTime = serviceObj.getLastRefTime();
if (!notifier.isSubscribed(serviceName, clusters) && !futureMap
.containsKey(ServiceInfo.getKey(serviceName, clusters))) {
// abort the update task
NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);
return;
}
if (CollectionUtils.isEmpty(serviceObj.getHosts())) {
//服务主机为空时，增加失败次数
incFailCount();
return;
}
delayTime = serviceObj.getCacheMillis();
//重置失败次数
resetFailCount();
} catch (Throwable e) {
incFailCount();
NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
} finally {
//失败次数参与计算下次服务更新任务延时时间，最大是60s，Math.min(delayTime << failCount, DEFAULT_DELAY * 60)
executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS);
}
}