Soulゲートウェイのソースコードは読みます。（12）http長ポーリング技術の詳細
第十編では、http長ポーリングを使ってデータを同期するという分析をしました。この記事は主に技術の詳細を分析します。余計なことを言わないで、直接ソースを入れます。
背景はHttpSyncDataServiceというデータ同期を行う実現クラスであり、構造関数を初期化する際に対象メソッドstartを呼び出した（）。まずスタート方法から分析に値する技術点を検討します。
コード断片1——start方法は、AtomicBoolean原子データタイプを使用して、同時スイッチ制御を処理し、サービスまたはスレッドのオンと停止が必要なときにこのように使用することができる。ここでは、synchronizdを使ってロックをかけても良いです。スイッチの操作は同期してスレッドが見えるだけでいいです。しかし、原子データの種類はcas技術を採用しているので、性能はさらに優れています。 はスレッド数とadminサービス数が等しいスレッドプールを作成し、スレッド工場をカスタマイズしてスレッドオブジェクトを作成する。作成したスレッドを守護スレッドに設定し、守護スレッドが非ガードスレッドで終了すると自動的に廃棄されます。 コードセグメント2——fetGroup Config excutor方式では、java 8を使用したストリーミングシンタックスparallel Streamをパラレルに処理したデータを更新していますが、全体の処理フローはブロックされています。つまり、parallel Stream.foreachの各ラインの間は並行していますが、successはストリーム処理の結果を同期します。ここでは、parramは非同期スレッド処理であるため、返ってくるデータはfinal配列を用いて受信されるが、ここでは、スレッドをまたいでデータ同期を行うために、finalオブジェクトの値を変更することによって同期を行わなければならない（対象の場所のヒープ空間はスレッド共有であるため。）、ベースデータタイプbollanの値はスタックに保存されているため、視認性を備えない。 コードセグメント3——ソウルエンドdoLongPolling処理ロジック fetGroupConfig方法において、私たちはすでに全量の構成データを引き出しました。次に、非同期タスクをスレッド池で実行して、各adminサービスに単独同期データを提供します。具体的には、run方法の論理です。ここでrun法はwhile輪訓の方式を採用し、一つのforサイクルを使って3回のdoLongPolling操作を行った。 DongPolling操作は、ローカルのキャッシュデータをmapに挿入してadmin/configs/listenerインターフェースに送信し、もう一度adminの構成データを取得したことが分かります。問題は何で3回doLongPolling操作を実行しますか？コードセグメント4を見る必要があります。adminサーバー/configs/listenerインターフェース論理 コードセグメント4——admin端doLongPolling処理ロジック compreCharedGroupはソウルゲートウェイから送られてきたデータを比較しました。簡単に言えば、データが一致しているかどうかを検出します。一致しない場合、新しいデータを生成し、ソウル要求側に直ちに同期します。 が異なるなら、servlet要求のコンテキストを非同期的に取得する。この特性はservlet 3.0バージョン以上が必要です。このようにする目的は、Servletコンテキストの情報を非同期スレッドに同期させて処理することをサポートし、非同期スレッドがrequest上下情報を取得できるようにすることである。（その技術原理を考えてみてください。） は、構造関数から、非同期スレッドが1つのコアスレッド数だけのタイミングスレッド実行器を使用して実行されることを見ることができる。具体的なタスクロジックは、run方法から、タイミングタスク実行器exectorを使用してタイミングタスクを実行し、次いで今回の呼び出し元の要求clientをブロック列に追加する（最大1024個）。タイムミッションの論理は60秒の遅延で実行されます。ブロック列から60秒前に入れたLongPollingClientオブジェクトを除去して、asyncConteextから再度compreChamedGroupを実行して、フルプロファイルを取得し、soul呼び出し側に送信します。ここでは、要求が頻繁すぎると、列がふさがってしまい、多くの要求が異常を投げてしまい、コントロールフローの効果が得られます。 コードセグメント5——compreCharedGroup SOul要求側のデータmd 5とclientModifyTime値 を取得する。は、adminサービスキャッシュのデータをmd 5値と要求側のmd 5値とを比較することにより、一致すればfalseに戻る。md 5の値が違っていたら、最後の修正時間を比較して、adminサーバーの修正時間がclientに等しい時間より大きいなら、再同期する必要がないと説明します。そうでなければ、同期adminサービス端末のデータをclientに送ります。 は、データ更新の同期性を保証するためにRentrantLockを使用して、併発問題を避ける。 締め括りをつける
このように、soul-webとadminの間では、純粋にhttpに基づくデータ同期が行われています。httpは全二重通信ではないので、クライアントにデータを自動的に押し付けることはできません。データの同期をクライアントが引く方式で行うと、データがリアルタイムに同期しない場合が必ずあります。httpに基づいてデータを同期させるには、一つのwhile輪訓＋要求時間間隔を制御するしかない。それだけではなく、多くの合併や異常を考慮する必要があります。は、失敗後の再送処理ポリシーを要求する。コードセグメント3——異常リピート クライアント要求が頻繁すぎて、サーバー側のフロー制御ポリシー。コードセグメント4——ブロック列を使用する サービス端末はどうやってデータを同期するかどうかを決定します。コードセグメント5——md 5+更新時間ダブルチェック
背景はHttpSyncDataServiceというデータ同期を行う実現クラスであり、構造関数を初期化する際に対象メソッドstartを呼び出した（）。まずスタート方法から分析に値する技術点を検討します。
コード断片1——start方法
private void start() {
// It could be initialized multiple times, so you need to control that.
if (RUNNING.compareAndSet(false, true)) {
//1
// 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)); //2
// start long polling, each server creates a thread to listen for changes.
this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));
} else {
log.info("soul http long polling was started, executor=[{}]", executor);
}
}
public static ThreadFactory create(final String namePrefix, final boolean daemon, final int priority) {
return new SoulThreadFactory(namePrefix, daemon, priority);
}
@Override
public Thread newThread(final Runnable runnable) {
Thread thread = new Thread(THREAD_GROUP, runnable,
THREAD_GROUP.getName() + "-" + namePrefix + "-" + THREAD_NUMBER.getAndIncrement());
thread.setDaemon(daemon);
thread.setPriority(priority);
return thread;
}
private void fetchGroupConfig(final ConfigGroupEnum... groups) throws SoulException {
for (int index = 0; index < this.serverList.size(); index++) {
String server = serverList.get(index);
try {
this.doFetchGroupConfig(server, groups);
break;
} catch (SoulException e) {
// no available server, throw exception.
if (index >= serverList.size() - 1) {
throw e;
}
log.warn("fetch config fail, try another one: {}", serverList.get(index + 1));
}
}
}
private void doFetchGroupConfig(final String server, final ConfigGroupEnum... groups) {
StringBuilder params = new StringBuilder();
for (ConfigGroupEnum groupKey : groups) {
params.append("groupKeys").append("=").append(groupKey.name()).append("&");
}
String url = server + "/configs/fetch?" + StringUtils.removeEnd(params.toString(), "&");
log.info("request configs: [{}]", url);
String json = null;
try {
json = this.httpClient.getForObject(url, String.class);
} catch (RestClientException e) {
String message = String.format("fetch config fail from server[%s], %s", url, e.getMessage());
log.warn(message);
throw new SoulException(message, e);
}
// update local cache
boolean updated = this.updateCacheWithJson(json);
if (updated) {
log.info("get latest configs: [{}]", json);
return;
}
// not updated. it is likely that the current config server has not been updated yet. wait a moment.
log.info("The config of the server[{}] has not been updated or is out of date. Wait for 30s to listen for changes again.", server);
ThreadUtils.sleep(TimeUnit.SECONDS, 30);
}
private boolean updateCacheWithJson(final String json) {
JsonObject jsonObject = GSON.fromJson(json, JsonObject.class);
JsonObject data = jsonObject.getAsJsonObject("data");
// if the config cache will be updated?
return factory.executor(data);
}
public boolean executor(final JsonObject data) {
final boolean[] success = {
false};
ENUM_MAP.values().parallelStream().forEach(dataRefresh -> success[0] = dataRefresh.refresh(data));//1
return success[0];
}
@Override
public void run() {
while (RUNNING.get()) {
for (int time = 1; time <= retryTimes; time++) {
//1
try {
doLongPolling(server);
} catch (Exception e) {
// print warnning log.
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;
}
// print error, then suspended for a while.
log.error("Long polling failed, try again after 5 minutes!", e);
ThreadUtils.sleep(TimeUnit.MINUTES, 5);
}
}
}
log.warn("Stop http long polling.");
}
@SuppressWarnings("unchecked")
private void doLongPolling(final String server) {
MultiValueMap<String, String> params = new LinkedMultiValueMap<>(8);
//2
for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
ConfigData<?> cacheConfig = factory.cacheConfigData(group);
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);
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) {
...
}
}
public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {
// compare group md5
List<ConfigGroupEnum> changedGroup = compareChangedGroup(request); //1
String clientIp = getRemoteIp(request);
// response immediately.
if (CollectionUtils.isNotEmpty(changedGroup)) {
this.generateResponse(response, changedGroup);
log.info("send response with the changed group, ip={}, group={}", clientIp, changedGroup);
return;
}
// listen for configuration changed.
final AsyncContext asyncContext = request.startAsync(); //2
// AsyncContext.settimeout() does not timeout properly, so you have to control it yourself
asyncContext.setTimeout(0L);
// block client's thread.
scheduler.execute(new LongPollingClient(asyncContext, clientIp, HttpConstants.SERVER_MAX_HOLD_TIMEOUT)); //3
}
public HttpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {
this.clients = new ArrayBlockingQueue<>(1024);
this.scheduler = new ScheduledThreadPoolExecutor(1,
SoulThreadFactory.create("long-polling", true));
this.httpSyncProperties = httpSyncProperties;
}
class LongPollingClient implements Runnable {
...
@Override
public void run() {
this.asyncTimeoutFuture = scheduler.schedule(() -> {
clients.remove(LongPollingClient.this);
List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
sendResponse(changedGroups);
}, timeoutTime, TimeUnit.MILLISECONDS);
clients.add(this);
}
...
}
private List<ConfigGroupEnum> compareChangedGroup(final HttpServletRequest request) {
List<ConfigGroupEnum> changedGroup = new ArrayList<>(ConfigGroupEnum.values().length);
for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
// md5,lastModifyTime
String[] params = StringUtils.split(request.getParameter(group.name()), ',');
if (params == null || params.length != 2) {
throw new SoulException("group param invalid:" + request.getParameter(group.name()));
}
String clientMd5 = params[0];
long clientModifyTime = NumberUtils.toLong(params[1]); //1
ConfigDataCache serverCache = CACHE.get(group.name());
// do check.
if (this.checkCacheDelayAndUpdate(serverCache, clientMd5, clientModifyTime)) {
changedGroup.add(group);
}
}
return changedGroup;
}
private boolean checkCacheDelayAndUpdate(final ConfigDataCache serverCache, final String clientMd5, final long clientModifyTime) {
// is the same, doesn't need to be updated
if (StringUtils.equals(clientMd5, serverCache.getMd5())) {
//2
return false;
}
// if the md5 value is different, it is necessary to compare lastModifyTime.
long lastModifyTime = serverCache.getLastModifyTime();
if (lastModifyTime >= clientModifyTime) {
// the client's config is out of date.
return true;
}
// the lastModifyTime before client, then the local cache needs to be updated.
// Considering the concurrency problem, admin must lock,
// otherwise it may cause the request from soul-web to update the cache concurrently, causing excessive db pressure
boolean locked = false;
try {
locked = LOCK.tryLock(5, TimeUnit.SECONDS); //3
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return true;
}
if (locked) {
try {
ConfigDataCache latest = CACHE.get(serverCache.getGroup());
if (latest != serverCache) {
// the cache of admin was updated. if the md5 value is the same, there's no need to update.
return !StringUtils.equals(clientMd5, latest.getMd5());
}
// load cache from db.
this.refreshLocalCache();
latest = CACHE.get(serverCache.getGroup());
return !StringUtils.equals(clientMd5, latest.getMd5());
} finally {
LOCK.unlock();
}
}
// not locked, the client need to be updated.
return true;
}
このように、soul-webとadminの間では、純粋にhttpに基づくデータ同期が行われています。httpは全二重通信ではないので、クライアントにデータを自動的に押し付けることはできません。データの同期をクライアントが引く方式で行うと、データがリアルタイムに同期しない場合が必ずあります。httpに基づいてデータを同期させるには、一つのwhile輪訓＋要求時間間隔を制御するしかない。それだけではなく、多くの合併や異常を考慮する必要があります。