Apache ShenYu 源码阅读系列 - 基于 Http 长轮询的数据同步

        <!--shenyu data sync start use http-->        <dependency>          <groupId>org.apache.shenyu</groupId>          <artifactId>shenyu-spring-boot-starter-sync-data-http</artifactId>          <version>${project.version}</version>        </dependency>

shenyu:    sync:       http:          url : http://localhost:9095

/** * Http sync data configuration for spring boot. */@Configuration@ConditionalOnClass(HttpSyncDataService.class)@ConditionalOnProperty(prefix = "shenyu.sync.http", name = "url")@Slf4jpublic class HttpSyncDataConfiguration {
    /**     * Http sync data service.     * 创建 HttpSyncDataService      * @param httpConfig         http的配置     * @param pluginSubscriber   插件数据订阅     * @param metaSubscribers    元数据订阅     * @param authSubscribers    认证数据订阅     * @return the sync data service     */    @Bean    public SyncDataService httpSyncDataService(final ObjectProvider<HttpConfig> httpConfig, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,                                           final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {        log.info("you use http long pull sync shenyu data");        return new HttpSyncDataService(Objects.requireNonNull(httpConfig.getIfAvailable()), Objects.requireNonNull(pluginSubscriber.getIfAvailable()),                metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));    }
    /**     * Http config http config.     * 读取http的配置     * @return the http config     */    @Bean    @ConfigurationProperties(prefix = "shenyu.sync.http")    public HttpConfig httpConfig() {        return new HttpConfig();    }}

public class HttpSyncDataService implements SyncDataService, AutoCloseable {
    // 省略了属性字段......
    public HttpSyncDataService(final HttpConfig httpConfig, final PluginDataSubscriber pluginDataSubscriber, final List<MetaDataSubscriber> metaDataSubscribers, final List<AuthDataSubscriber> authDataSubscribers) {        // 1.创建数据处理器        this.factory = new DataRefreshFactory(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);        // 2.获取admin属性配置        this.httpConfig = httpConfig;        // shenyu-admin的url， 多个用逗号(,)分割        this.serverList = Lists.newArrayList(Splitter.on(",").split(httpConfig.getUrl()));        // 3.创建httpClient，用于向admin发起请求        this.httpClient = createRestTemplate();        // 4.开始执行长轮询任务        this.start();    }
    //......}

      private RestTemplate createRestTemplate() {          OkHttp3ClientHttpRequestFactory factory = new OkHttp3ClientHttpRequestFactory();            // 建立连接超时时间为 10s          factory.setConnectTimeout((int) this.connectionTimeout.toMillis());            // 网关主动请求 shenyu-admin 的配置服务，读取超时时间为 90s          factory.setReadTimeout((int) HttpConstants.CLIENT_POLLING_READ_TIMEOUT);          return new RestTemplate(factory);      }

private void start() {        // 只初始化一次，通过原子类实现。         RUNNING = new AtomicBoolean(false);        // 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<>(),                    ShenyuThreadFactory.create("http-long-polling", true));            // start long polling, each server creates a thread to listen for changes.                        // 开始长轮询，一个admin服务，创建一个线程用于数据同步            this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));        } else {            log.info("shenyu http long polling was started, executor=[{}]", executor);        }    }

public enum ConfigGroupEnum {    APP_AUTH, // 认证数据    PLUGIN, //插件    RULE, // 规则    SELECTOR, // 选择器    META_DATA; // 元数据}

private void fetchGroupConfig(final ConfigGroupEnum... groups) throws ShenyuException {    // admin有可能是集群，这里通过循环的方式向每个admin发起请求        for (int index = 0; index < this.serverList.size(); index++) {            String server = serverList.get(index);            try {                // 真正去执行                this.doFetchGroupConfig(server, groups);                // 有一个成功，就成功了，可以退出循环                break;            } catch (ShenyuException 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));            }        }    }

// 向admin后台管理系统发起请求，获取所有同步数据private void doFetchGroupConfig(final String server, final ConfigGroupEnum... groups) {    // 1. 拼请求参数，所有分组枚举类型    StringBuilder params = new StringBuilder();    for (ConfigGroupEnum groupKey : groups) {        params.append("groupKeys").append("=").append(groupKey.name()).append("&");    }
    // admin端提供的接口  /configs/fetch    String url = server + "/configs/fetch?" + StringUtils.removeEnd(params.toString(), "&");    log.info("request configs: [{}]", url);    String json = null;    try {        // 2. 发起请求，获取变更数据        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 ShenyuException(message, e);    }    // update local cache    // 3. 更新网关内存中数据    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);    // 服务端没有数据更新，就等30s    ThreadUtils.sleep(TimeUnit.SECONDS, 30);}

    private boolean updateCacheWithJson(final String json) {        // 使用GSON进行反序列化        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) {        //并行更新数据        List<Boolean> result = ENUM_MAP.values().parallelStream()                .map(dataRefresh -> dataRefresh.refresh(data))                .collect(Collectors.toList());        //有一个更新就表示此次发生了更新操作        return result.stream().anyMatch(Boolean.TRUE::equals);    }

    @Override    public Boolean refresh(final JsonObject data) {        boolean updated = false;        // 数据类型转换        JsonObject jsonObject = convert(data);        if (null != jsonObject) {            // 得到数据类型            ConfigData<T> result = fromJson(jsonObject);            // 是否需要更新            if (this.updateCacheIfNeed(result)) {                updated = true;                // 真正的更新逻辑，数据刷新操作                refresh(result.getData());            }        }        return updated;    }

// result是数据protected abstract boolean updateCacheIfNeed(ConfigData<T> result);
// newVal是获取到的最新的值// groupEnum 是哪种数据类型protected boolean updateCacheIfNeed(final ConfigData<T> newVal, final ConfigGroupEnum groupEnum) {        // 如果是第一次，那么直接放到cache中，返回 true，表示此次进行了更新        if (GROUP_CACHE.putIfAbsent(groupEnum, newVal) == null) {            return true;        }        ResultHolder holder = new ResultHolder(false);        GROUP_CACHE.merge(groupEnum, newVal, (oldVal, value) -> {            // md5 值相同，不需要更新            if (StringUtils.equals(oldVal.getMd5(), newVal.getMd5())) {                log.info("Get the same config, the [{}] config cache will not be updated, md5:{}", groupEnum, oldVal.getMd5());                return oldVal;            }
            // 当前缓存的数据修改时间大于 新来的数据，不需要更新            // must compare the last update time            if (oldVal.getLastModifyTime() >= newVal.getLastModifyTime()) {                log.info("Last update time earlier than the current configuration, the [{}] config cache will not be updated", groupEnum);                return oldVal;            }            log.info("update {} config: {}", groupEnum, newVal);            holder.result = true;            return newVal;        });        return holder.result;    }

    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<>(),                    ShenyuThreadFactory.create("http-long-polling", true));            // start long polling, each server creates a thread to listen for changes.            // 开始长轮询，一个admin服务，创建一个线程用于数据同步            this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));        } else {            log.info("shenyu http long polling was started, executor=[{}]", executor);        }    }

class HttpLongPollingTask implements Runnable {
        private String server;
        // 默认重试 3 次        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) {                        // 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());                            // 长轮询失败了，等 5s 再继续                            ThreadUtils.sleep(TimeUnit.SECONDS, 5);                            continue;                        }                        // print error, then suspended for a while.                        log.error("Long polling failed, try again after 5 minutes!", e);                        // 3 次都失败了，等 5 分钟再试                        ThreadUtils.sleep(TimeUnit.MINUTES, 5);                    }                }            }            log.warn("Stop http long polling.");        }    }

private void doLongPolling(final String server) {        // 组装请求参数：md5 和 lastModifyTime        MultiValueMap<String, String> params = new LinkedMultiValueMap<>(8);        for (ConfigGroupEnum group : ConfigGroupEnum.values()) {            ConfigData<?> cacheConfig = factory.cacheConfigData(group);            if (cacheConfig != null) {                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;        //向admin发起请求，判断组数据是否发生变更        //这里只是判断了某个组是否发生变更        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 ShenyuException(message, e);        }        // 根据发生变更的组，再去获取数据        /**         * 官网对此处的解释：         * 网关收到响应信息之后，只知道是哪个 Group 发生了配置变更，还需要再次请求该 Group 的配置数据。         * 这里可能会存在一个疑问：为什么不是直接将变更的数据写出？         * 我们在开发的时候，也深入讨论过该问题，因为 http 长轮询机制只能保证准实时，如果在网关层处理不及时，         * 或者管理员频繁更新配置，很有可能便错过了某个配置变更的推送，安全起见，我们只告知某个 Group 信息发生了变更。         *         * 个人理解：         * 如果将变更数据直接写出，当管理员频繁更新配置时，第一次更新了，将client移除阻塞队列，返回响应信息给网关。         * 如果这个时候进行了第二次更新，那么当前的client是不在阻塞队列中，所以这一次的变更就会错过。         * 网关层处理不及时，也是同理。         * 这是一个长轮询，一个网关一个同步线程，可能存在耗时的过程。         * 如果admin有数据变更，当前网关client是没有在阻塞队列中，就不到数据。         */        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));                // 主动向admin获取变更的数据，根据分组不同，全量拿数据                this.doFetchGroupConfig(server, changedGroups);            }        }    }

shenyu:  sync:      http:        enabled: true

/** * 数据同步配置类 * 通过springboot条件装配实现 * The type Data sync configuration. */@Configurationpublic class DataSyncConfiguration {
    /**     * http长轮询     * http long polling.     */    @Configuration    @ConditionalOnProperty(name = "shenyu.sync.http.enabled", havingValue = "true")    @EnableConfigurationProperties(HttpSyncProperties.class)    static class HttpLongPollingListener {
        @Bean        @ConditionalOnMissingBean(HttpLongPollingDataChangedListener.class)        public HttpLongPollingDataChangedListener httpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {            return new HttpLongPollingDataChangedListener(httpSyncProperties);        }    }}

    public HttpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {        // 默认客户端（这里是网关）1024个        this.clients = new ArrayBlockingQueue<>(1024);        // 创建线程池        // ScheduledThreadPoolExecutor 可以执行延迟任务，周期任务，普通任务        this.scheduler = new ScheduledThreadPoolExecutor(1,                ShenyuThreadFactory.create("long-polling", true));        // 长轮询的属性信息        this.httpSyncProperties = httpSyncProperties;    }

    /**     * 在 InitializingBean接口中的afterPropertiesSet()方法中被调用，即在bean的初始化过程中执行     */    @Override    protected void afterInitialize() {        long syncInterval = httpSyncProperties.getRefreshInterval().toMillis();        // Periodically check the data for changes and update the cache
        // 执行周期任务：更新内存中（CACHE）的数据每隔5分钟执行一次，5分钟后开始执行        // 防止admin先启动一段时间后，产生了数据；然后网关初次连接时，没有拿到全量数据        scheduler.scheduleWithFixedDelay(() -> {            log.info("http sync strategy refresh config start.");            try {                // 从数据库读取数据到本地缓存（这里就是内存）                this.refreshLocalCache();                log.info("http sync strategy refresh config success.");            } catch (Exception e) {                log.error("http sync strategy refresh config error!", e);            }        }, syncInterval, syncInterval, TimeUnit.MILLISECONDS);        log.info("http sync strategy refresh interval: {}ms", syncInterval);    }

    // 从数据库读取数据到本地缓存（这里就是内存）    private void refreshLocalCache() {        //更新认证数据        this.updateAppAuthCache();        //更新插件数据        this.updatePluginCache();        //更新规则数据        this.updateRuleCache();        //更新选择器数据        this.updateSelectorCache();        //更新元数据        this.updateMetaDataCache();    }

    /**     * Update rule cache.     */    protected void updateRuleCache() {        this.updateCache(ConfigGroupEnum.RULE, ruleService.listAll());    }

// 缓存数据的 Mapprotected static final ConcurrentMap<String, ConfigDataCache> CACHE = new ConcurrentHashMap<>();
/**     * if md5 is not the same as the original, then update lcoal cache.     * 更新缓存中的数据     * @param group ConfigGroupEnum     * @param <T> the type of class     * @param data the new config data     */    protected <T> void updateCache(final ConfigGroupEnum group, final List<T> data) {        //数据序列化        String json = GsonUtils.getInstance().toJson(data);        //传入md5值和修改时间        ConfigDataCache newVal = new ConfigDataCache(group.name(), json, Md5Utils.md5(json), System.currentTimeMillis());        //更新分组数据        ConfigDataCache oldVal = CACHE.put(newVal.getGroup(), newVal);        log.info("update config cache[{}], old: {}, updated: {}", group, oldVal, newVal);    }

   /** * This Controller only when HttpLongPollingDataChangedListener exist, will take effect. */@ConditionalOnBean(HttpLongPollingDataChangedListener.class)@RestController@RequestMapping("/configs")@Slf4jpublic class ConfigController {
    @Resource    private HttpLongPollingDataChangedListener longPollingListener;        // 省略其他逻辑
    /**     * Listener.     * 监听数据变更，执行长轮询     * @param request  the request     * @param response the response     */    @PostMapping(value = "/listener")    public void listener(final HttpServletRequest request, final HttpServletResponse response) {        longPollingListener.doLongPolling(request, response);    }
}

/**     * 执行长轮询：如果有数据变更，会立即响应给客户端（这里就是网关端）。     * 否则，否则客户端会一直被阻塞，直到有数据变更或者超时。     * @param request     * @param response     */    public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {        // compare group md5        // 比较md5，判断网关的数据和admin端的数据是否一致，得到发生变更的数据组        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));    }

 /**     * 判断组数据是否发生变更     * @param request     * @return     */    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 ShenyuException("group param invalid:" + request.getParameter(group.name()));            }            String clientMd5 = params[0];            long clientModifyTime = NumberUtils.toLong(params[1]);            ConfigDataCache serverCache = CACHE.get(group.name());            // do check. 判断组数据是否发生变更            if (this.checkCacheDelayAndUpdate(serverCache, clientMd5, clientModifyTime)) {                changedGroup.add(group);            }        }        return changedGroup;    }

class LongPollingClient implements Runnable {      // 省略了其他逻辑            @Override        public void run() {            try {                // 60秒后移除，并响应客户端                this.asyncTimeoutFuture = scheduler.schedule(() -> {                    clients.remove(LongPollingClient.this);                    List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());                    sendResponse(changedGroups);                }, timeoutTime, TimeUnit.MILLISECONDS);
                // 添加到阻塞队列                clients.add(this);
            } catch (Exception ex) {                log.error("add long polling client error", ex);            }        }
        /**         * Send response.         *         * @param changedGroups the changed groups         */        void sendResponse(final List<ConfigGroupEnum> changedGroups) {            // cancel scheduler            if (null != asyncTimeoutFuture) {                asyncTimeoutFuture.cancel(false);            }            // 响应变更的组            generateResponse((HttpServletResponse) asyncContext.getResponse(), changedGroups);            asyncContext.complete();        }    }

@ConditionalOnBean(HttpLongPollingDataChangedListener.class)@RestController@RequestMapping("/configs")@Slf4jpublic class ConfigController {
    @Resource    private HttpLongPollingDataChangedListener longPollingListener;
    /**     * Fetch configs shenyu result.     * 全量获取分组数据     * @param groupKeys the group keys     * @return the shenyu result     */    @GetMapping("/fetch")    public ShenyuAdminResult 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 ShenyuAdminResult.success(ShenyuResultMessage.SUCCESS, result);    }      // 省略了其他接口
}

    /**     * fetch configuration from cache.     * 获取分组下的全量数据     * @param groupKey the group key     * @return the configuration data     */    public ConfigData<?> fetchConfig(final ConfigGroupEnum groupKey) {        // 直接从 CACHE 中拿数据        ConfigDataCache config = CACHE.get(groupKey.name());         switch (groupKey) {            case APP_AUTH: // 认证数据                List<AppAuthData> appAuthList = GsonUtils.getGson().fromJson(config.getJson(), new TypeToken<List<AppAuthData>>() {                }.getType());                return new ConfigData<>(config.getMd5(), config.getLastModifyTime(), appAuthList);            case PLUGIN: // 插件数据                List<PluginData> pluginList = GsonUtils.getGson().fromJson(config.getJson(), new TypeToken<List<PluginData>>() {                }.getType());                return new ConfigData<>(config.getMd5(), config.getLastModifyTime(), pluginList);            case RULE:   // 规则数据                List<RuleData> ruleList = GsonUtils.getGson().fromJson(config.getJson(), new TypeToken<List<RuleData>>() {                }.getType());                return new ConfigData<>(config.getMd5(), config.getLastModifyTime(), ruleList);            case SELECTOR:  // 选择器数据                List<SelectorData> selectorList = GsonUtils.getGson().fromJson(config.getJson(), new TypeToken<List<SelectorData>>() {                }.getType());                return new ConfigData<>(config.getMd5(), config.getLastModifyTime(), selectorList);            case META_DATA: // 元数据                List<MetaData> metaList = GsonUtils.getGson().fromJson(config.getJson(), new TypeToken<List<MetaData>>() {                }.getType());                return new ConfigData<>(config.getMd5(), config.getLastModifyTime(), metaList);            default:  // 其他类型，抛出异常                throw new IllegalStateException("Unexpected groupKey: " + groupKey);        }

/** * Event forwarders, which forward the changed events to each ConfigEventListener. * 数据变更事件分发器：当admin端有数据发生变更时，将变更数据同步到 ShenYu 网关 * 数据变更依赖于Spring的事件监听机制：ApplicationEventPublisher --> ApplicationEvent --> ApplicationListener * */@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
   //省略了其他逻辑
    /**     * 有数据变更时，调用此方法     * @param event     */    @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // 遍历数据变更监听器(一般使用一种数据同步的方式就好了)        for (DataChangedListener listener : listeners) {            // 哪种数据发生变更            switch (event.getGroupKey()) {                case APP_AUTH: // 认证信息                    listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());                    break;                case PLUGIN:  // 插件信息                    listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());                    break;                case RULE:    // 规则信息                    listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());                    break;                case SELECTOR:   // 选择器信息                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());                    break;                case META_DATA:  // 元数据                    listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());                    break;                default:  // 其他类型，抛出异常                    throw new IllegalStateException("Unexpected value: " + event.getGroupKey());            }        }    }}

    /**     * 在admin的操作，有插件发生了更新     * @param changed   the changed     * @param eventType the event type     */    @Override    public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {        if (CollectionUtils.isEmpty(changed)) {            return;        }        // 更新内存CACHE        this.updatePluginCache();        // 执行变更任务        this.afterPluginChanged(changed, eventType);    }

    @Override    protected void afterPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {        // 在线程池中执行        scheduler.execute(new DataChangeTask(ConfigGroupEnum.PLUGIN));    }

class DataChangeTask implements Runnable {    //省略了其他逻辑         @Override        public void run() {            // 阻塞队列中的客户端超过了给定的值100，则分批执行            if (clients.size() > httpSyncProperties.getNotifyBatchSize()) {                List<LongPollingClient> targetClients = new ArrayList<>(clients.size());                clients.drainTo(targetClients);                List<List<LongPollingClient>> partitionClients = Lists.partition(targetClients, httpSyncProperties.getNotifyBatchSize());               // 分批执行                partitionClients.forEach(item -> scheduler.execute(() -> doRun(item)));            } else {                // 执行任务                doRun(clients);            }        }
        private void doRun(final Collection<LongPollingClient> clients) {            // 通知所有客户端发生了数据变更            for (Iterator<LongPollingClient> iter = clients.iterator(); iter.hasNext();) {                LongPollingClient client = iter.next();                iter.remove();                // 发送响应                client.sendResponse(Collections.singletonList(groupKey));                log.info("send response with the changed group,ip={}, group={}, changeTime={}", client.ip, groupKey, changeTime);            }        }    }