One post tagged with "http"

/**
 * Http sync data configuration for spring boot.
 */
@Configuration
@ConditionalOnClass(HttpSyncDataService.class)
@ConditionalOnProperty(prefix = "shenyu.sync.http", name = "url")
@EnableConfigurationProperties(value = HttpConfig.class)
public class HttpSyncDataConfiguration {

  private static final Logger LOGGER = LoggerFactory.getLogger(HttpSyncDataConfiguration.class);

  /**
   * Rest template.
   * 创建RestTemplate
   * @param httpConfig the http config       http配置
   * @return the rest template
   */
  @Bean
  public RestTemplate restTemplate(final HttpConfig httpConfig) {
    OkHttp3ClientHttpRequestFactory factory = new OkHttp3ClientHttpRequestFactory();
    factory.setConnectTimeout(Objects.isNull(httpConfig.getConnectionTimeout()) ? (int) HttpConstants.CLIENT_POLLING_CONNECT_TIMEOUT : httpConfig.getConnectionTimeout());
    factory.setReadTimeout(Objects.isNull(httpConfig.getReadTimeout()) ? (int) HttpConstants.CLIENT_POLLING_READ_TIMEOUT : httpConfig.getReadTimeout());
    factory.setWriteTimeout(Objects.isNull(httpConfig.getWriteTimeout()) ? (int) HttpConstants.CLIENT_POLLING_WRITE_TIMEOUT : httpConfig.getWriteTimeout());
    return new RestTemplate(factory);
  }

  /**
   * AccessTokenManager.
   * 创建AccessTokenManager,专门用户对admin进行http请求时access token的处理
   * @param httpConfig   the http config.      
   * @param restTemplate the rest template.
   * @return the access token manager.
   */
  @Bean
  public AccessTokenManager accessTokenManager(final HttpConfig httpConfig, final RestTemplate restTemplate) {
    return new AccessTokenManager(restTemplate, httpConfig);
  }

  /**
   * Http sync data service.
   * 创建 HttpSyncDataService 
   * @param httpConfig         the http config
   * @param pluginSubscriber   the plugin subscriber
   * @param restTemplate       the rest template
   * @param metaSubscribers    the meta subscribers
   * @param authSubscribers    the auth subscribers
   * @param accessTokenManager the access token manager
   * @return the sync data service
   */
  @Bean
  public SyncDataService httpSyncDataService(final ObjectProvider<HttpConfig> httpConfig,
                                             final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
                                             final ObjectProvider<RestTemplate> restTemplate,
                                             final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers,
                                             final ObjectProvider<List<AuthDataSubscriber>> authSubscribers,
                                             final ObjectProvider<AccessTokenManager> accessTokenManager) {
    LOGGER.info("you use http long pull sync shenyu data");
    return new HttpSyncDataService(
            Objects.requireNonNull(httpConfig.getIfAvailable()),
            Objects.requireNonNull(pluginSubscriber.getIfAvailable()),
            Objects.requireNonNull(restTemplate.getIfAvailable()),
            metaSubscribers.getIfAvailable(Collections::emptyList),
            authSubscribers.getIfAvailable(Collections::emptyList),
            Objects.requireNonNull(accessTokenManager.getIfAvailable())
    );
  }
}

public class HttpSyncDataService implements SyncDataService {

    // 省略了属性字段......

    public HttpSyncDataService(final HttpConfig httpConfig,
                               final PluginDataSubscriber pluginDataSubscriber,
                               final RestTemplate restTemplate,
                               final List<MetaDataSubscriber> metaDataSubscribers,
                               final List<AuthDataSubscriber> authDataSubscribers,
                               final AccessTokenManager accessTokenManager) {
          // 1.设置accessTokenManager
          this.accessTokenManager = accessTokenManager;
          // 2.创建数据处理器
          this.factory = new DataRefreshFactory(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);
          // 3.shenyu-admin的url， 多个用逗号(,)分割
          this.serverList = Lists.newArrayList(Splitter.on(",").split(httpConfig.getUrl()));
          // 4.只用于http长轮询的restTemplate
          this.restTemplate = restTemplate;
          // 5.开始执行长轮询任务
          this.start();
    }

    //......
}

public class HttpSyncDataService implements SyncDataService {
    
    // ......

    private void start() {
      // // 只初始化一次，通过原子类实现。 
      if (RUNNING.compareAndSet(false, true)) {
        // 初次启动，获取全量数据
        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));
        // 开始长轮询，一个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; // 元数据
}

public class HttpSyncDataService implements SyncDataService {

  // ......

  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) {
        // 出现异常，尝试执行下一个
        // 最后一个也执行失败了，抛出异常
        if (index >= serverList.size() - 1) {
          throw e;
        }
        LOG.warn("fetch config fail, try another one: {}", serverList.get(index + 1));
      }
    }
  }

  // ......
}

public class HttpSyncDataService implements SyncDataService {
  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 + Constants.SHENYU_ADMIN_PATH_CONFIGS_FETCH + "?" + StringUtils.removeEnd(params.toString(), "&");
    LOG.info("request configs: [{}]", url);
    String json;
    try {
      HttpHeaders headers = new HttpHeaders();
      // 设置accessToken
      headers.set(Constants.X_ACCESS_TOKEN, this.accessTokenManager.getAccessToken());
      HttpEntity<String> httpEntity = new HttpEntity<>(headers);
      // 2. 发起请求，获取变更数据
      json = this.restTemplate.exchange(url, HttpMethod.GET, httpEntity, String.class).getBody();
    } 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);
    }
    // 3. 更新网关内存中数据
    boolean updated = this.updateCacheWithJson(json);
    if (updated) {
      LOG.debug("get latest configs: [{}]", json);
      return;
    }
    // 更新成功，此方法就执行完成了
    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);
        // if the config cache will be updated?
        return factory.executor(jsonObject.getAsJsonObject("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);
    }

public abstract class AbstractDataRefresh<T> implements DataRefresh {

  // ......

  @Override
  public Boolean refresh(final JsonObject data) {
    // 数据类型转换
    JsonObject jsonObject = convert(data);
    if (Objects.isNull(jsonObject)) {
      return false;
    }

    boolean updated = false;
    // 得到数据类型
    ConfigData<T> result = fromJson(jsonObject);
    // 是否需要更新
    if (this.updateCacheIfNeed(result)) {
      updated = true;
      // 真正的更新逻辑，数据刷新操作
      refresh(result.getData());
    }

    return updated;
  }

  // ......
}

public abstract class AbstractDataRefresh<T> implements DataRefresh {

  // ......

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

  // ......
}

public class HttpSyncDataService implements SyncDataService {

  // ......

  private void start() {
    // // 只初始化一次，通过原子类实现。 
    if (RUNNING.compareAndSet(false, true)) {
      // 初次启动，获取全量数据
      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));
      // 开始长轮询，一个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 final String server;

  HttpLongPollingTask(final String server) {
    this.server = server;
  }

  @Override
  public void run() {
    // 一直轮询
    while (RUNNING.get()) {
      // 默认重试 3 次
      int retryTimes = 3;
      for (int time = 1; time <= retryTimes; time++) {
        try {
          doLongPolling(server);
        } catch (Exception e) {
          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;
          }
          LOG.error("Long polling failed, try again after 5 minutes!", e);
          // 3 次都失败了，等 5 分钟再试
          ThreadUtils.sleep(TimeUnit.MINUTES, 5);
        }
      }
    }
    LOG.warn("Stop http long polling.");
  }
}

public class HttpSyncDataService implements SyncDataService {
  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);
    // 设置accessToken
    headers.set(Constants.X_ACCESS_TOKEN, this.accessTokenManager.getAccessToken());
    HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(params, headers);
    String listenerUrl = server + Constants.SHENYU_ADMIN_PATH_CONFIGS_LISTENER;

    JsonArray groupJson;
    //向admin发起请求，判断组数据是否发生变更
    //这里只是判断了某个组是否发生变更
    try {
      String json = this.restTemplate.postForEntity(listenerUrl, httpEntity, String.class).getBody();
      LOG.info("listener result: [{}]", json);
      JsonObject responseFromServer = GsonUtils.getGson().fromJson(json, JsonObject.class);
      groupJson = responseFromServer.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 (Objects.nonNull(groupJson) && groupJson.size() > 0) {
      // fetch group configuration async.
      ConfigGroupEnum[] changedGroups = GsonUtils.getGson().fromJson(groupJson, ConfigGroupEnum[].class);
      LOG.info("Group config changed: {}", Arrays.toString(changedGroups));
      this.doFetchGroupConfig(server, changedGroups);
    }
  }
}

/**
 * 数据同步配置类
 * 通过springboot条件装配实现
 * The type Data sync configuration.
 */
@Configuration
public 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;
    }

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {

  // ......

  /**
   * 在 InitializingBean接口中的afterPropertiesSet()方法中被调用，即在bean的初始化过程中执行
   */
  @Override
  protected void afterInitialize() {
    long syncInterval = httpSyncProperties.getRefreshInterval().toMillis();
    // 执行周期任务：更新内存中（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);
  }

  // ......
}

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {

  // ......

  // 从数据库读取数据到本地缓存（这里就是内存）
  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());
    }

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {

  // ......

  // 缓存数据的 Map
  protected 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")
public class ConfigController {

    private final HttpLongPollingDataChangedListener longPollingListener;

    public ConfigController(final HttpLongPollingDataChangedListener longPollingListener) {
      this.longPollingListener = 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);
    }

}

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {

  // ......

  /**
   * 执行长轮询：如果有数据变更，会立即响应给客户端（这里就是网关端）。
   * 否则，否则客户端会一直被阻塞，直到有数据变更或者超时。
   * @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;
    }

    // 没有变更，则将客户端（这里就是网关）放进阻塞队列
    final AsyncContext asyncContext = request.startAsync();
    asyncContext.setTimeout(0L);
    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")
public class ConfigController {

    private final HttpLongPollingDataChangedListener longPollingListener;
  
    public ConfigController(final HttpLongPollingDataChangedListener longPollingListener) {
      this.longPollingListener = 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);
    }
    
  // 省略了其他接口

}

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {
  /**
   * 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: // 认证数据
        return buildConfigData(config, AppAuthData.class);
      case PLUGIN: // 插件数据
        return buildConfigData(config, PluginData.class);
      case RULE:   // 规则数据
        return buildConfigData(config, RuleData.class);
      case SELECTOR:  // 选择器数据
        return buildConfigData(config, SelectorData.class);
      case META_DATA: // 元数据
        return buildConfigData(config, MetaData.class);
      default:  // 其他类型，抛出异常
        throw new IllegalStateException("Unexpected groupKey: " + groupKey);
    }
  }
}

/**
 * Event forwarders, which forward the changed events to each ConfigEventListener.
 * 数据变更事件分发器：当admin端有数据发生变更时，将变更数据同步到 ShenYu 网关
 * 数据变更依赖于Spring的事件监听机制：ApplicationEventPublisher --> ApplicationEvent --> ApplicationListener
 *
 */
@Component
public 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());
                    // 当选择器数据更新时，更新API文档信息
                    applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnSelectorChanged((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);
            }
        }
    }