@Before
public void init() throws NacosException {
Properties properties = new Properties();
properties.put(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:" + 8848);
properties.put(PropertyKeyConst.CONFIG_LONG_POLL_TIMEOUT, "20000");
properties.put(PropertyKeyConst.CONFIG_RETRY_TIME, "3000");
properties.put(PropertyKeyConst.MAX_RETRY, "5");
configService = NacosFactory.createConfigService(properties);
}
@Test
public void test() throws InterruptedException, NacosException {
configService.addListener("test", "DEFAULT_GROUP", new Listener() {
@Override
public Executor getExecutor() {
return null;
}
@Override
public void receiveConfigInfo(String configInfo) {
System.out.println(configInfo);
}
});
TimeUnit.SECONDS.sleep(100);
}
这是Nacos给客户端提供的API,可以通过该API:增、删、盖、查配置信息,还可以通过该API给配置添加Listener
// NacosFactory#createConfigService
public static ConfigService createConfigService(Properties properties) throws NacosException {
return ConfigFactory.createConfigService(properties);
}
// ConfigFactory#createConfigService
public static ConfigService createConfigService(Properties properties) throws NacosException {
try {
Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.config.NacosConfigService");
Constructor constructor = driverImplClass.getConstructor(Properties.class);
// 反射创建对象
ConfigService vendorImpl = (ConfigService) constructor.newInstance(properties);
return vendorImpl;
} catch (Throwable e) {
throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e);
}
}
通过反射的机制创建了一个NacosConfigService
实例,它的构造函数如下
public NacosConfigService(Properties properties) throws NacosException {
String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE);
if (StringUtils.isBlank(encodeTmp)) {
encode = Constants.ENCODE;
} else {
encode = encodeTmp.trim();
}
initNamespace(properties);
agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
agent.start();
worker = new ClientWorker(agent, configFilterChainManager, properties);
}
MetricsHttpAgent
主要是用来记录一些Metric信息,内部持有一个ServerHttpAgent
对象,所以主要逻辑还是关注ServerHttpAgent
;ClientWorker
是一个非常核心的类,里面封装了客户端从服务端主动PULL配置信息的关键逻辑,这个在下面重点介绍public ServerHttpAgent(Properties properties) throws NacosException {
serverListMgr = new ServerListManager(properties);
init(properties);
}
ServerHttpAgent
内部主要就是封装了一个HTTP交互的通用方法,比如GET
,PUT
,DELETE
等方法,在ServerHttpAgent
构造函数中,主要做了两件事情:
ServerListManager
对象,这个对象的主要作用就是根据Properties
解析出服务端的地址,然后维护在一个List<String>
中init
方法中主要做了三件事情:初始化编码格式,如果没有就默认UTF-8;初始化accessKey
和secretKey
,这个应该是用来验证客户端的身份的;初始化重试次数,如果没传默认为3在ServerListManager
对象中,还有一个比较重要的属性isFixed
,这个属性的主要作用就是标识服务器的地址信息是不是固定的,比如,如果服务器的地址是通过Properties
传入,那isFixed
的值为true
为什么要介绍这个属性呢,因为在NacosConfigService
的构造函数中,调用了MetricsHttpAgent#start
方法,而这个方法的内部调用链如下
MetricsHttpAgent#start =>
ServerHttpAgent#start =>
ServerListManager#start
ServerListManager#start
方法主要做了什么事情呢?如果isFixed
的值为true,就直接返回。否则先执行initServerlistRetryTimes
次GetServerListTask#run
方法获取,该方法用于更新服务器地址信息,如果执行initServerlistRetryTimes
次
之后还是没有获取到服务器地址列表信息,则直接抛出异常,否则开启一个定时任务,每30s更新一次服务器地址列表信息。
至此,在NacosConfigService
构造函数中,只剩下ClientWorker
相关的逻辑了
前面已经提到过,ClientWorker
是一个非常核心的类,里面封装了客户端从服务端主动pull
配置信息的关键逻辑。注意这里很明确指出了ClientWorker
是通过pull模式
从服务端获取配置信息的,而我们在使用的时候通常会给它添加Listener
,这
会让我们以为它是push模式
,这是一点需要注意的地方。
而它实现的pull模式
的关键点在于两个线程池,这两个线程池在ClientWorker
的构造函数中初始化。其中有一个线程池executor
里面只有一个线程,它的作用就是让另一个线程池启动。
public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager, final Properties properties) {
init(properties);
executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
t.setDaemon(true);
return t;
}
});
executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker.longPolling." + agent.getName());
t.setDaemon(true);
return t;
}
});
executor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
checkConfigInfo();
} catch (Throwable e) {
LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e);
}
}
}, 1L, 10L, TimeUnit.MILLISECONDS);
}
init(properties)
方法主要就是初始化一些timeout的参数executor线程池
里面只有一个线程,它的唯一作用就是让另一个线程池开始执行,每10s中执行一次executorService线程池
用于更新配置信息,核心任务LongPollingRunnable
ClientWorker#checkConfigInfo
方法主要作用是更新配置信息,目前已经获取到的配置信息会缓存到一个Map<String, CacheData>
中,然后对map中的数据分批次,一个批次默认是3000条数据,每个批次的数据对应一个线程负责更新,如下:
public void checkConfigInfo() {
// 分任务
int listenerSize = cacheMap.get().size();
// 向上取整为批数
int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
if (longingTaskCount > currentLongingTaskCount) {
for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
// 要判断任务是否在执行 这块需要好好想想。 任务列表现在是无序的。变化过程可能有问题
executorService.execute(new LongPollingRunnable(i));
}
currentLongingTaskCount = longingTaskCount;
}
}
从LongPollingRunnable
的名字可以看出,客户端主要是通过长轮询的方式去更新配置信息
public void run() {
// 本批次号的数据
List<CacheData> cacheDatas = new ArrayList<CacheData>();
// 本批次号中 isInitializing=true 的数据,CacheData首次出现在Map中并且是首次check更新时,isInitializing的值才为true
List<String> inInitializingCacheList = new ArrayList<String>();
try {
// check failover config
for (CacheData cacheData : cacheMap.get().values()) {
if (cacheData.getTaskId() == taskId) {
cacheDatas.add(cacheData);
try {
// 涉及到FailoverFile
checkLocalConfig(cacheData);
// 如果有更新,需要更新Listener的MD5值,并执行Listener逻辑
if (cacheData.isUseLocalConfigInfo()) {
cacheData.checkListenerMd5();
}
} catch (Exception e) {
LOGGER.error("get local config info error", e);
}
}
}
// check server config
// 找到isUseLocalConfig=false的数据,然后将每个符合条件的CacheData的 `group + dataId + tenant` 拼成一个字符串传给服务端校验,然后服务端会返回一个需要更新的`List<String>`,该列表里面的每个元素代表一个CacheData的key
List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
for (String groupKey : changedGroupKeys) {
String[] key = GroupKey.parseKey(groupKey);
String dataId = key[0];
String group = key[1];
String tenant = null;
if (key.length == 3) {
tenant = key[2];
}
try {
// 根据需要更新的key列表,从服务端获取配置信息
String content = getServerConfig(dataId, group, tenant, 3000L);
CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
cache.setContent(content);
} catch (NacosException ioe) {
}
}
// 这部分代码是在没看懂
for (CacheData cacheData : cacheDatas) {
if (!cacheData.isInitializing() || inInitializingCacheList.contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) {
cacheData.checkListenerMd5();
cacheData.setInitializing(false);
}
}
inInitializingCacheList.clear();
// 重新提交自己
executorService.execute(this);
} catch (Throwable e) {
// If the rotation training task is abnormal, the next execution time of the task will be punished
LOGGER.error("longPolling error : ", e);
executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
}
}
LongPollingRunnable#run
方法比较长:
CacheData
都维护了一个批次号信息,然后每个LongPollingRunnable
也对应一个批次号信息,只需要负责更新批次号相同的数据
,不同的由别的LongPollingRunnable
去更新,这部分据放在变量cacheDatas
中isUseLocalConfig=true
,并更新CacheData
的值,同时更新CacheData
的MD5值isUseLocalConfig==true
,此时执行cacheData#checkListenerMd5
方法,该方法会拿CacheData
的MD5值和Listener
的MD5值对比,如果不一样就更新Listener
的MD5值并回调Listener
的相关方法isUseLocalConfig=true
的配置信息,然后将每个符合条件的配置信息的 group + dataId +tenant
拼成一个字符串传给服务端校验,然后服务端会返回一个需要更新的List<String>
,该列表里面的每个元素代表一个CacheData
的keyCacheData
的值cacheData.isInitializing()
代表该条配置信息首次出现在map中并且是首次更新,inInitializingCacheList
代表本批次中isUseLocalConfig==false && isInitializing == true
isInitializing == false || isInitializing == true
?上面提到ClientWorker
中有一个Map,那里面的数据从哪里来呢?为什么要存这部分数据呢?答案是在为CacheData
添加Listener
的时候,会初始化一条CacheData
数据,并添加到Map中
ConfigService configService = NacosFactory.createConfigService(properties);
configService.addListener("test", "DEFAULT_GROUP", new Listener() {
@Override
public Executor getExecutor() {
return null;
}
@Override
public void receiveConfigInfo(String configInfo) {
System.out.println(configInfo);
}
});
源码
// NacosConfigService#addListener
public void addListener(String dataId, String group, Listener listener) throws NacosException {
worker.addTenantListeners(dataId, group, Arrays.asList(listener));
}
// ClientWorker#addTenantListeners
public void addTenantListeners(String dataId, String group, List<? extends Listener> listeners) throws NacosException {
group = null2defaultGroup(group);
String tenant = agent.getTenant();
CacheData cache = addCacheDataIfAbsent(dataId, group, tenant);
for (Listener listener : listeners) {
cache.addListener(listener);
}
}
内部还是委托给ClientWorker
来实现,添加监时主要做了以下几件事情
dataId group tenant
封装成key,从ClientWorker
的Map缓存中获取CacheData
1.1. 如果从缓存中拿到了数据,将CacheData
的isInitializing
属性设置为true
1.2. 如果从缓存中没拿到数据,则创建一个CacheData
,同时判断是否开启远程同步,即enableRemoteSyncConfig
属性的值是否为true。如果开启了,则从服务端获取该配置的值,并设置到CacheData
中CacheData
设置到缓存中Listener
添加到CacheData
的CopyOnWriteArrayList<ManagerListenerWrap> listeners
列表中其实在上面已经提到过,当配置有更新时,会触发Listener
的回调逻辑,这部分逻辑在CacheData#checkListenerMd5
方法中
void checkListenerMd5() {
// `ManagerListenerWrap`只是一个包装类,里面维护了`Listener`和对应`CacheData`的MD5值
for (ManagerListenerWrap wrap : listeners) {
// 如果不一样,说明配置有变更,此时更新ManagerListenerWrap的MD5值,并执行Listener的回调
if (!md5.equals(wrap.lastCallMd5)) {
safeNotifyListener(dataId, group, content, md5, wrap);
}
}
}
private void safeNotifyListener(final String dataId, final String group, final String content, final String md5, final ManagerListenerWrap listenerWrap) {
final Listener listener = listenerWrap.listener;
Runnable job = new Runnable() {
@Override
public void run() {
try {
ConfigResponse cr = new ConfigResponse();
cr.setDataId(dataId);
cr.setGroup(group);
cr.setContent(content);
configFilterChainManager.doFilter(null, cr);
String contentTmp = cr.getContent();
listener.receiveConfigInfo(contentTmp);
listenerWrap.lastCallMd5 = md5;
} finally {
Thread.currentThread().setContextClassLoader(myClassLoader);
}
}
};
final long startNotify = System.currentTimeMillis();
try {
if (null != listener.getExecutor()) {
// 如果配置了线程池,交给线程池执行
listener.getExecutor().execute(job);
} else {
job.run();
}
} catch (Throwable t) {
}
}
ManagerListenerWrap
只是一个包装类,里面维护了Listener
和对应CacheData
的MD5值ManagerListenerWrap
和当前CacheData
的MD5值是否相同,如果不一样,说明配置有变更,此时需要更新ManagerListenerWrap
的MD5值,并执行Listener
的回调ManagerListenerWrap
的MD5值和执行Listener
的回调的逻辑都在safeNotifyListener
方法中,同时会判断是否为Listener
配置了线程池如,没有就直接执行,有就交给线程池执行ConfigService configService = NacosFactory.createConfigService(properties);
configService.publishConfig("one", "LSZ", "大美女");
源码
private boolean publishConfigInner(String tenant, String dataId, String group, String tag, String appName,
String betaIps, String content) throws NacosException {
...... 一大坨构造参数的代码 ......
HttpResult result = null;
try {
// 推送到服务端
result = agent.httpPost(url, headers, params, encode, POST_TIMEOUT);
} catch (IOException ioe) {
return false;
}
......
}
ConfigService configService = NacosFactory.createConfigService(properties);
configService.getConfig("testlsz", "lszgroup", 1000000);
源码
private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {
......
// 优先使用本地配置
String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
if (content != null) {
return content;
}
try {
// 本地没有从服务器获取
content = worker.getServerConfig(dataId, group, tenant, timeoutMs);
cr.setContent(content);
configFilterChainManager.doFilter(null, cr);
content = cr.getContent();
return content;
} catch (NacosException ioe) {
}
......
}
ConfigService configService = NacosFactory.createConfigService(properties);
configService.removeConfig("testlsz", "lszgroup");
源码
private boolean removeConfigInner(String tenant, String dataId, String group, String tag) throws NacosException {
...... 一大坨构造参数的代码 ......
HttpResult result = null;
try {
// 直接发送http请求
result = agent.httpDelete(url, null, params, encode, POST_TIMEOUT);
} catch (IOException ioe) {
LOGGER.warn("[remove] error, " + dataId + ", " + group + ", " + tenant + ", msg: " + ioe.toString());
return false;
}
......
}
在Naco中涉及到两个文件,FailoverFile
和SnapshotFile
FailoverFile
为容灾文件,当本地和数据库里面数据不一致的时候会去使用,一般不会用;SnapshotFile
为配置的快照,当获取不到服务器上的配置的时候,会读取本地快照;
FailoverFile
在客户端不会自动生成,它是在服务端生成的,当更新了一条配置之后,就会反应到这个文件中。所以如果想在客户端使用到这个功能,需要手工将文件添加到客户端,然后客户端就不会去读取服务端的配置了,也许某些场景下可以用到SnapshotFile
文件位置:\userhome\nacos\config\fixed-127.0.0.1_8848_nacos\snapshot\{group}\{dataId}
当客户端从服务端获取配置之后,会将该信息写入快照文件中,核心代码就在ClientWorker#getServerConfig
中
public String getServerConfig(String dataId, String group, String tenant, long readTimeout) throws NacosException {
HttpResult result = null;
try {
// 从服务端获取配置信息
result = agent.httpGet(Constants.CONFIG_CONTROLLER_PATH, null, params, agent.getEncode(), readTimeout);
} catch (IOException e) {
throw new NacosException(NacosException.SERVER_ERROR, e);
}
switch (result.code) {
case HttpURLConnection.HTTP_OK:
// 将配置更新到本地文件
LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, result.content);
return result.content;
case HttpURLConnection.HTTP_NOT_FOUND:
// 根据 dataId、group、tenant 将本地的文件删除
LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, null);
return null;
case HttpURLConnection.HTTP_CONFLICT: {
throw new NacosException(NacosException.CONFLICT,"data being modified, dataId=" + dataId + ", group=" + group + ",tenant=" + tenant);
}
case HttpURLConnection.HTTP_FORBIDDEN: {
LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", agent.getName(), dataId, group, tenant);
throw new NacosException(result.code, result.content);
}
default: {
LOGGER.error("[{}] [sub-server-error] dataId={}, group={}, tenant={}, code={}", agent.getName(), dataId, group, tenant, result.code);
throw new NacosException(result.code, "http error, code=" + result.code + ",dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
}
}
}
FailoverFile
文件位置:\userhome\nacos\config\fixed-127.0.0.1_8848_nacos\data\config-data\{group}\{dataId}
对FailoverFile
文件的判断主要是在ClientWorker#checkLocalConfig
方法,看了几篇文章,都说这个方法的校验是为了在服务端挂了的时候,可以直接从客户端获取文件,这明显是不对的。刚刚在上面也提到过,FailoverFile
的主要作用是容灾,而且这个文件在客户端不会自动生成,想要使用这个功能必须手动添加
private void checkLocalConfig(CacheData cacheData) {
final String dataId = cacheData.dataId;
final String group = cacheData.group;
final String tenant = cacheData.tenant;
// 注意这个,容灾文件,这个文件在客户端是不会自动生成的
File path = LocalConfigInfoProcessor.getFailoverFile(agent.getName(), dataId, group, tenant);
// 没有 -> 有
if (!cacheData.isUseLocalConfigInfo() && path.exists()) {
String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
String md5 = MD5.getInstance().getMD5String(content);
cacheData.setUseLocalConfigInfo(true);
cacheData.setLocalConfigInfoVersion(path.lastModified());
cacheData.setContent(content);
LOGGER.warn("[{}] [failover-change] failover file created. dataId={}, group={}, tenant={}, md5={}, content={}",
agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));
return;
}
// 有 -> 没有。不通知业务监听器,从server拿到配置后通知。
if (cacheData.isUseLocalConfigInfo() && !path.exists()) {
cacheData.setUseLocalConfigInfo(false);
LOGGER.warn("[{}] [failover-change] failover file deleted. dataId={}, group={}, tenant={}", agent.getName(),
dataId, group, tenant);
return;
}
// 有变更
/**
* isUseLocalConfig=true && && path.exists() && cacheData.getLocalConfigInfoVersion() != path.lastModified()
* 说明配置有更新,所以此时需要更新 CacheData
*/
if (cacheData.isUseLocalConfigInfo() && path.exists()
&& cacheData.getLocalConfigInfoVersion() != path.lastModified()) {
String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
String md5 = MD5.getInstance().getMD5String(content);
cacheData.setUseLocalConfigInfo(true);
cacheData.setLocalConfigInfoVersion(path.lastModified());
cacheData.setContent(content);
LOGGER.warn("[{}] [failover-change] failover file changed. dataId={}, group={}, tenant={}, md5={}, content={}",
agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));
}
}
从上面已经知道,有一个线程池会不停的执行LongPollingRunnable#run
方法,这个方法主要作用就是从服务端拉取最新的配置信息,如果直接按照正常的做法来做,直接根据dataId group tenant
去拉取就好了,如果每次都直接去服务器来配置信息,但这样会有一些性能问题:
针对上面几个问题,Nacos做了以下几个优化
key列表
给服务端,服务端返回发生了变更的Key列表
,大部分时候,这可以过滤掉绝大部分没有配置项;先推荐一篇文章:Nacos配置实时更新原理分析 这篇文章已经写的非常详细了,不过那篇文章有点长,这里总结一下,为了自己以后看的时候方便。
通过HTTP长轮询较少客户端和服务端的交互频率,但这必然要面对一个数据响应不实时问问题,怎么解决?
在客户端向服务端拉取配置信息之前,需要先向服务端发送一个配置Key列表
,然后服务端返回一个发生了变更的配置Key列表
// timeout默认30s超时
HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params, agent.getEncode(), timeout);
请求的服务端地址: /v1/cs/configs/listener
@RequestMapping(value = "/listener", method = RequestMethod.POST)
public void listener(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
....
// do long-polling
inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}
// ConfigServletInner#doPollingConfig ,暂且只关注长轮询
// 长轮询
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) {
int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
/**
* 提前500ms返回响应,为避免客户端超时 add delay time for LoadBalance
*/
long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
String ip = RequestUtil.getRemoteIp(req);
// 一定要由HTTP线程调用,否则离开后容器会立即发送响应, 开启Servlet异步支持
final AsyncContext asyncContext = req.startAsync();
// AsyncContext.setTimeout()的超时时间不准,所以只能自己控制
asyncContext.setTimeout(0L);
// 向线程池提交了一个任务,所以核心逻辑在 ClientLongPolling 中,
scheduler.execute(new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}
public void run() {
asyncTimeoutFuture = scheduler.schedule(new Runnable() {
@Override
public void run() {
try {
getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
/**
* 删除订阅关系
*/
allSubs.remove(ClientLongPolling.this);
if (isFixedPolling()) {
// 根据客户端传过来的key列表,找到该发生了变更的配置的key列表
List<String> changedGroups = MD5Util.compareMd5((HttpServletRequest)asyncContext.getRequest(), (HttpServletResponse)asyncContext.getResponse(), clientMd5Map);
if (changedGroups.size() > 0) {
// 返回给客户端
sendResponse(changedGroups);
} else {
// 没有就返回空
sendResponse(null);
}
} else {
sendResponse(null);
}
} catch (Throwable t) {
LogUtil.defaultLog.error("long polling error:" + t.getMessage(), t.getCause());
}
}
}, timeoutTime, TimeUnit.MILLISECONDS);
allSubs.add(this);
}
ClientLongPolling
添加到allSubs
中ClientLongPolling
从allSubs
中移除,然后通过AsyncContext
将结果写回客户端allSubs
数据结构如下,可以把allSubs
当作是一个订阅者列表,当配置发生变成的时候,会发布一个事件,然后这些订阅者会得到相应,然后再执行相应的功能
Queue<ClientLongPolling> allSubs = new ConcurrentLinkedQueue<ClientLongPolling>();
延迟 29.5s 执行,要是在这段时间内,数据发生了变更怎么办,难道要客户端 29.5s 之后才知道吗?肯定不是的。可以看看当数据发生变更时,会涉及到什么接口
请求的服务端地址: /v1/cs/configs
// 持久化
persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, false);
// 触发事件
EventDispatcher.fireEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
static public void fireEvent(Event event) {
for (AbstractEventListener listener : getEntry(event.getClass()).listeners) {
try {
listener.onEvent(event);
} catch (Exception e) {
log.error(e.toString(), e);
}
}
}
static public abstract class AbstractEventListener {
public AbstractEventListener() {
EventDispatcher.addEventListener(this);
}
}
// EventDispatcher#addEventListener
static public void addEventListener(AbstractEventListener listener) {
for (Class<? extends Event> type : listener.interest()) {
getEntry(type).listeners.addIfAbsent(listener);
}
}
注意观察AbstractEventListener
的构造函数,在其构造函数中涉及到Listener
的注册过程,而AbstractEventListener
有以下几个子类:
AsyncNotifyService -> ConfigDataChangeEvent
LongPollingService -> LocalDataChangeEvent
MockListener 计数用的
刚刚看上面发布的是一个ConfigDataChangeEvent
事件,所以会先执行AsyncNotifyService#onEvent
方法。该方法中会先获取服务端所有IP列表,依次通过Http通知对象/v1/cs/communication/dataChange?dataId=xx&group=xx,接收到请求后,
会dump出来所有config信息,同时回调LocalDataChangeEvent事件,然后执行LongPollingService#onEvent
方法。
public void onEvent(Event event) {
if (isFixedPolling()) {
} else {
if (event instanceof LocalDataChangeEvent) {
LocalDataChangeEvent evt = (LocalDataChangeEvent)event;
scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
}
}
}
提交了一个任务,关键逻辑在DataChangeTask#run
方法中
public void run() {
try {
ConfigService.getContentBetaMd5(groupKey);
for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
// ClientLongPolling 作为订阅者,配置类似于Topic,更新配置就是发布了一个事件
ClientLongPolling clientSub = iter.next();
// 找到订阅了 当前发生了变更配置项 的ClientLongPolling
if (clientSub.clientMd5Map.containsKey(groupKey)) {
getRetainIps().put(clientSub.ip, System.currentTimeMillis());
// 删除订阅关系
iter.remove();
// 向客户端回写数据
clientSub.sendResponse(Arrays.asList(groupKey));
}
}
} catch (Throwable t) {
LogUtil.defaultLog.error("data change error:" + t.getMessage(), t.getCause());
}
}
这里会有一个问题,如果DataChangeTask
任务完成了向客户端写数据,此时ClientLongPolling
中的调度任务又开始执行了怎么办呢?这时任务都被移除了,肯定会报错啊
很简单,只要在进行"推送"操作之前,先将原来等待执行的调度任务取消掉就可以了,如下:
void sendResponse(List<String> changedGroups) {
/**
* 取消超时任务
*/
if (null != asyncTimeoutFuture) {
asyncTimeoutFuture.cancel(false);
}
generateResponse(changedGroups);
}
这样,就达到了一个类似于数据"推送"的效果,如果一直没有更新,客户端等待时间接近 30s,如果在等待期间有数据发生变更,几乎可以实时的返回给客户端