Soul 源码阅读 05｜Http 长轮询同步数据分析

    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));            // 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);        }    }

private void doLongPolling(final String server) {  MultiValueMap<String, String> params = new LinkedMultiValueMap<>(8);  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) {    String message = String.format("listener configs fail, server:[%s], %s", server, e.getMessage());    throw new SoulException(message, e);  }  if (groupJson != null) {    // fetch group configuration async.    ConfigGroupEnum[] changedGroups = GSON.fromJson(groupJson, ConfigGroupEnum[].class);    if (ArrayUtils.isNotEmpty(changedGroups)) {      log.info("Group config changed: {}", Arrays.toString(changedGroups));      this.doFetchGroupConfig(server, changedGroups);    }  }}

    @GetMapping("/fetch")    public SoulAdminResult fetchConfigs(@NotNull final String[] groupKeys) {        Map<String, ConfigData<?>> result = Maps.newHashMap();        for (String groupKey : groupKeys) {            ConfigData<?> data = longPollingListener.fetchConfig(ConfigGroupEnum.valueOf(groupKey));            result.put(groupKey, data);        }        return SoulAdminResult.success(SoulResultMessage.SUCCESS, result);    }

/**     * If the configuration data changes, the group information for the change is immediately responded.     * Otherwise, the client's request thread is blocked until any data changes or the specified timeout is reached.     *     * @param request  the request     * @param response the response     */public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {
  // compare group md5  List<ConfigGroupEnum> changedGroup = compareChangedGroup(request);  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();
  // 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));}