上一節(jié)我們講解了HttpSyncDataService在初始化后，start的fetch流程碳默，接下來贾陷，我們在看下他的定時監(jiān)聽流程。
private void start() {
// It could be initialized multiple times, so you need to control that.
if (RUNNING.compareAndSet(false, true)) {
// fetch all group configs.
this.fetchGroupConfig(ConfigGroupEnum.values());
int threadSize = serverList.size();
//初始化一個線程池
this.executor = new ThreadPoolExecutor(threadSize, threadSize, 60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
SoulThreadFactory.create("http-long-polling", true));
// 循環(huán)所有的server地址嘱根，創(chuàng)建HttpLongPollingTask異步執(zhí)行長輪訓(xùn)任務(wù)
this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));
} else {
log.info("soul http long polling was started, executor=[{}]", executor);
}
}
我們看下HttpLongPollingTask髓废，主要是循環(huán)的根據(jù)RUNNING狀態(tài)判斷是否需要進(jìn)行調(diào)用
//org.dromara.soul.sync.data.http.HttpSyncDataService.HttpLongPollingTask
class HttpLongPollingTask implements Runnable {
private String server;
private final int retryTimes = 3;
HttpLongPollingTask(final String server) {
this.server = server;
}
@Override
public void run() {
while (RUNNING.get()) {
for (int time = 1; time <= retryTimes; time++) {
try {
doLongPolling(server);
} catch (Exception e) {
// 如果拉取服務(wù)器配置的時候報錯，則重試该抒，最多重試3次慌洪，并等待5秒鐘
if (time < retryTimes) {
log.warn("Long polling failed, tried {} times, {} times left, will be suspended for a while! {}",
time, retryTimes - time, e.getMessage());
ThreadUtils.sleep(TimeUnit.SECONDS, 5);
continue;
}
重試超過3次后，報錯凑保，并等待5分鐘
log.error("Long polling failed, try again after 5 minutes!", e);
ThreadUtils.sleep(TimeUnit.MINUTES, 5);
}
}
}
log.warn("Stop http long polling.");
}
}
doLongPolling這個方法用到了我們之前在SoulAdmin看到的/configs/listener
//org.dromara.soul.sync.data.http.HttpSyncDataService#doLongPolling
private void doLongPolling(final String server) {
MultiValueMap<String, String> params = new LinkedMultiValueMap<>(8);
//循環(huán)所有的ConfigGroup冈爹，并從之前fetch拉取到的數(shù)據(jù)中獲取本地的配置數(shù)據(jù)
for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
ConfigData<?> cacheConfig = factory.cacheConfigData(group);
//構(gòu)造參數(shù)value值
String value = String.join(",", cacheConfig.getMd5(), String.valueOf(cacheConfig.getLastModifyTime()));
params.put(group.name(), Lists.newArrayList(value));
}
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
HttpEntity httpEntity = new HttpEntity(params, headers);
//http調(diào)用SoulAdmin的configs/listener
String listenerUrl = server + "/configs/listener";
log.debug("request listener configs: [{}]", listenerUrl);
JsonArray groupJson = null;
try {
String json = this.httpClient.postForEntity(listenerUrl, httpEntity, String.class).getBody();
log.debug("listener result: [{}]", json);
groupJson = GSON.fromJson(json, JsonObject.class).getAsJsonArray("data");
} catch (RestClientException e) {
String message = String.format("listener configs fail, server:[%s], %s", server, e.getMessage());
throw new SoulException(message, e);
}
//獲取到結(jié)果的json
if (groupJson != null) {
// 發(fā)現(xiàn)哪些是需要變化的group
ConfigGroupEnum[] changedGroups = GSON.fromJson(groupJson, ConfigGroupEnum[].class);
if (ArrayUtils.isNotEmpty(changedGroups)) {
//再通過fetch拉取對應(yīng)group的最新數(shù)據(jù)
log.info("Group config changed: {}", Arrays.toString(changedGroups));
this.doFetchGroupConfig(server, changedGroups);
}
}
}
到這里我們看到了，HttpSyncDataService在start的時候欧引，剛開始會全量fetch频伤，并啟動一個異步線程，對所有的SoulAdmin繼續(xù)進(jìn)行異步的listener芝此。
我們總結(jié)一下流程圖：
image.png
到這里我們將SoulAdmin配置相關(guān)同步機制都講完了
