2 posts 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.     *     * @param httpConfig the http config     * @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.     *     * @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.     *     * @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 {
    // omitted attribute field ......
    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. create data refresh factory          this.factory = new DataRefreshFactory(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);          // 3. shenyu-admin url          this.serverList = Lists.newArrayList(Splitter.on(",").split(httpConfig.getUrl()));          // 4. restTemplate          this.restTemplate = restTemplate;          // 5. start a long polling task          this.start();    }
    //......}

public class HttpSyncDataService implements SyncDataService {
    // ......
    private void start() {        // It could be initialized multiple times, so you need to control that.        if (RUNNING.compareAndSet(false, true)) {            // fetch all group configs.            // Initial startup, get full data            this.fetchGroupConfig(ConfigGroupEnum.values());            // one backend service, one thread            int threadSize = serverList.size();            // ThreadPoolExecutor            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.            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, // app auth data    PLUGIN, // plugin data    RULE, // rule data    SELECTOR, // selector data    META_DATA; // meta data}

public class HttpSyncDataService implements SyncDataService {
    // ......
    private void fetchGroupConfig(final ConfigGroupEnum... groups) throws ShenyuException {        // It is possible that admins are clustered, and here requests are made to each admin by means of a loop.        for (int index = 0; index < this.serverList.size(); index++) {            String server = serverList.get(index);            try {                // do execute                this.doFetchGroupConfig(server, groups);                // If you have a success, you are successful and can exit the loop                break;            } catch (ShenyuException e) {                // An exception occurs, try executing the next                // The last one also failed to execute, throwing an exception                // no available server, throw exception.                if (index >= serverList.size() - 1) {                    throw e;                }                LOG.warn("fetch config fail, try another one: {}", serverList.get(index + 1));            }        }    }
    // ......}

public class HttpSyncDataService implements SyncDataService {
    // ......
    // Launch a request to the admin backend management system to get all synchronized data    private void doFetchGroupConfig(final String server, final ConfigGroupEnum... groups) {        // 1. build request parameters, all grouped enumeration types        StringBuilder params = new StringBuilder();        for (ConfigGroupEnum groupKey : groups) {            params.append("groupKeys").append("=").append(groupKey.name()).append("&");        }        // admin url:  /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();            // set accessToken            headers.set(Constants.X_ACCESS_TOKEN, this.accessTokenManager.getAccessToken());            HttpEntity<String> httpEntity = new HttpEntity<>(headers);            // 2. get a request for change data            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);        }        // update local cache        // 3. Update data in gateway memory        boolean updated = this.updateCacheWithJson(json);        if (updated) {            LOG.debug("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);        // No data update on the server side, just wait 30s        ThreadUtils.sleep(TimeUnit.SECONDS, 30);    }
    // ......}

public class HttpSyncDataService implements SyncDataService {
    // ......
    private boolean updateCacheWithJson(final String json) {        // Using GSON for deserialization        JsonObject jsonObject = GSON.fromJson(json, JsonObject.class);        // if the config cache will be updated?        return factory.executor(jsonObject.getAsJsonObject("data"));    }
    // ......}

public final class DataRefreshFactory {        // ......        public boolean executor(final JsonObject data) {        // update data        List<Boolean> result = ENUM_MAP.values().parallelStream()                .map(dataRefresh -> dataRefresh.refresh(data))                .collect(Collectors.toList());        // one of the data types is updated, which means that the operation has been updated.        return result.stream().anyMatch(Boolean.TRUE::equals);    }        // ......}

public abstract class AbstractDataRefresh<T> implements DataRefresh {
    // ......
    @Override    public Boolean refresh(final JsonObject data) {        // convert data        JsonObject jsonObject = convert(data);        if (Objects.isNull(jsonObject)) {            return false;        }
        boolean updated = false;        // get data        ConfigData<T> result = fromJson(jsonObject);        // does it need to be updated        if (this.updateCacheIfNeed(result)) {            updated = true;            // real update logic, data refresh operation            refresh(result.getData());        }
        return updated;    }
    // ......}

public abstract class AbstractDataRefresh<T> implements DataRefresh {
    // ......
    // result is data    protected abstract boolean updateCacheIfNeed(ConfigData<T> result);
    // newVal is the latest value obtained    // What kind of data type is groupEnum    protected boolean updateCacheIfNeed(final ConfigData<T> newVal, final ConfigGroupEnum groupEnum) {        // If it is the first time, then it is put directly into the cache and returns true, indicating that the update was made this time        if (GROUP_CACHE.putIfAbsent(groupEnum, newVal) == null) {            return true;        }        ResultHolder holder = new ResultHolder(false);        GROUP_CACHE.merge(groupEnum, newVal, (oldVal, value) -> {            // md5 value is the same, no need to update            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;            }
            // The current cached data has been modified for a longer period than the new data and does not need to be updated.            // 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() {        // It could be initialized multiple times, so you need to control that.        if (RUNNING.compareAndSet(false, true)) {            // fetch all group configs.            // Initial startup, get full data            this.fetchGroupConfig(ConfigGroupEnum.values());            // one backend service, one thread            int threadSize = serverList.size();            // ThreadPoolExecutor            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.            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() {        // long polling        while (RUNNING.get()) {            // Default retry 3 times            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());                        // long polling failed, wait 5s and continue                        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.");    }}

public class HttpSyncDataService implements SyncDataService {    private void doLongPolling(final String server) {        // build request params: md5 and 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));            }        }        // build request head and body        HttpHeaders headers = new HttpHeaders();        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);        // set 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;        //Initiate a request to admin to determine if the group data has changed        //Here it just determines whether a group has changed or not        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);        }        // Depending on the group where the change occurred, go back and get the data        /**         * The official website explains here.         * After the gateway receives the response message, it only knows which Group has made the configuration change, and it still needs to request the configuration data of that Group again.         * There may be a question here: why not write out the changed data directly?         * We also discussed this issue in depth during development, because the http long polling mechanism can only guarantee quasi-real time, if the processing at the gateway layer is not timely, * or the administrator frequently updates the configuration, it is very difficult to get the information from the gateway layer.         * If it is not processed in time at the gateway level, or if the administrator updates the configuration frequently, it is very likely to miss the push of a configuration change, so for security reasons, we only inform a group that the information has changed.         *For security reasons, we only notify a group of changes.         * Personal understanding.         * If the change data is written out directly, when the administrator frequently updates the configuration, the first update will remove the client from the blocking queue and return the response information to the gateway.         * If a second update is made at this time, the current client is not in the blocking queue, so this time the change is missed.         * The same is true for untimely processing by the gateway layer.         * This is a long polling, one gateway one synchronization thread, there may be time consuming process.         * If the admin has data changes, the current gateway client is not in the blocking queue and will not get the data.         */        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));                // Proactively get the changed data from admin, depending on the grouping, and take the data in full                this.doFetchGroupConfig(server, changedGroups);            }        }        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));            // Proactively get the changed data from admin, depending on the grouping, and take the data in full            this.doFetchGroupConfig(server, changedGroups);        }    }}

/** * Data synchronization configuration class * Conditional assembly via springboot * The type Data sync configuration. */@Configurationpublic class DataSyncConfiguration {
    /**     * 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) {        // default client (here is the gateway) 1024        this.clients = new ArrayBlockingQueue<>(1024);        // create thread pool        // ScheduledThreadPoolExecutor can perform delayed tasks, periodic tasks, and normal tasks        this.scheduler = new ScheduledThreadPoolExecutor(1,                ShenyuThreadFactory.create("long-polling", true));        // http sync properties        this.httpSyncProperties = httpSyncProperties;    }

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {
    // ......
    /**     * is called in the afterPropertiesSet() method of the InitializingBean interface, which is executed during the initialization of the bean     */    @Override    protected void afterInitialize() {        long syncInterval = httpSyncProperties.getRefreshInterval().toMillis();        // Periodically check the data for changes and update the cache
        // Execution cycle task: Update data in memory (CACHE) is executed every 5 minutes and starts after 5 minutes        // Prevent the admin from starting up first for a while and then generating data; then the gateway doesn't get the full amount of data when it first connects        scheduler.scheduleWithFixedDelay(() -> {            LOG.info("http sync strategy refresh config start.");            try {                // Read data from database to local cache (in this case, memory)                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 {
    // ......
    // Read data from database to local cache (in this case, memory)    private void refreshLocalCache() {        //update app auth data        this.updateAppAuthCache();        //update plugin data        this.updatePluginCache();        //update rule data        this.updateRuleCache();        //update selector data        this.updateSelectorCache();        //update meta data        this.updateMetaDataCache();    }
    // ......}

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

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {
    // ......
    // cache 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) {        // data serialization        String json = GsonUtils.getInstance().toJson(data);        // pass in md5 value and modification time        ConfigDataCache newVal = new ConfigDataCache(group.name(), json, Md5Utils.md5(json), System.currentTimeMillis());        // update group data        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;    }        // Omit other logic
    /**     * Listener.     * Listen for data changes and perform long polling     * @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 {
    // ......
    /**     * Execute long polling: If there is a data change, it will be responded to the client (here is the gateway side) immediately.     * Otherwise, the client will otherwise remain blocked until there is a data change or a timeout.     * @param request     * @param response     */    public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {        // compare group md5        // Compare the md5, determine whether the data of the gateway and the data of the admin side are consistent, and get the data group that has changed        List<ConfigGroupEnum> changedGroup = compareChangedGroup(request);        String clientIp = getRemoteIp(request);        // response immediately.        // Immediate response to the gateway if there is changed data        if (CollectionUtils.isNotEmpty(changedGroup)) {            this.generateResponse(response, changedGroup);            Log.info("send response with the changed group, ip={}, group={}", clientIp, changedGroup);            return;        }
        // No change, then the client (in this case the gateway) is put into the blocking queue        // 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));    }}

 /**     * Determine if the group data has changed     * @param request     * @return     */    private List<ConfigGroupEnum> compareChangedGroup(final HttpServletRequest request) {        List<ConfigGroupEnum> changedGroup = new ArrayList<>(ConfigGroupEnum.values().length);        for (ConfigGroupEnum group : ConfigGroupEnum.values()) {            // The md5 value and lastModifyTime of the data on the gateway side            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. determine if the group data has changed            if (this.checkCacheDelayAndUpdate(serverCache, clientMd5, clientModifyTime)) {                changedGroup.add(group);            }        }        return changedGroup;    }

class LongPollingClient implements Runnable {      // omitted other logic            @Override        public void run() {            try {                // Removal after 60 seconds and response to the client                this.asyncTimeoutFuture = scheduler.schedule(() -> {                    clients.remove(LongPollingClient.this);                    List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());                    sendResponse(changedGroups);                }, timeoutTime, TimeUnit.MILLISECONDS);
                // Add to blocking queue                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);            }            // Groups responding to changes            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);    }      // Other interfaces are omitted
}

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) {        // get data from CACHE        ConfigDataCache config = CACHE.get(groupKey.name());        switch (groupKey) {            case APP_AUTH: // app auth data                return buildConfigData(config, AppAuthData.class);            case PLUGIN: // plugin data                return buildConfigData(config, PluginData.class);            case RULE:   // rule data                return buildConfigData(config, RuleData.class);            case SELECTOR:  // selector data                return buildConfigData(config, SelectorData.class);            case META_DATA: // meta data                 return buildConfigData(config, MetaData.class);            default:  // other data type, throw exception                throw new IllegalStateException("Unexpected groupKey: " + groupKey);        }    }        // ......}

/** * Event forwarders, which forward the changed events to each ConfigEventListener. * Data change event distributor: synchronize the change data to ShenYu gateway when there is a data change in admin side * Data changes rely on Spring's event-listening mechanism: ApplicationEventPublisher --> ApplicationEvent --> ApplicationListener * */@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
  // other logic omitted
    /**     * Call this method when there are data changes     * @param event     */    @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // Iterate through the data change listeners (it's generally good to use a kind of data synchronization)        for (DataChangedListener listener : listeners) {            // What kind of data has changed            switch (event.getGroupKey()) {                case APP_AUTH: // app auth data                    listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());                    break;                case PLUGIN:  // plugin data                    listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());                    break;                case RULE:    // rule data                    listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());                    break;                case SELECTOR:   // selector data                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());                    // pull and save API document on seletor changed                    applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());                    break;                case META_DATA:  // meta data                    listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());                    break;                default:  // other data type, throw exception                    throw new IllegalStateException("Unexpected value: " + event.getGroupKey());            }        }    }}

    /**     * In the operation of the admin, there is an update of the plugin occurred     * @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;        }        // update CACHE        this.updatePluginCache();        // execute change task        this.afterPluginChanged(changed, eventType);    }

    @Override    protected void afterPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {        // execute by thread pool        scheduler.execute(new DataChangeTask(ConfigGroupEnum.PLUGIN));    }

class DataChangeTask implements Runnable {        //other logic omitted          @Override        public void run() {            // If the client in the blocking queue exceeds the given value of 100, it is executed in batches            if (clients.size() > httpSyncProperties.getNotifyBatchSize()) {                List<LongPollingClient> targetClients = new ArrayList<>(clients.size());                clients.drainTo(targetClients);                List<List<LongPollingClient>> partitionClients = Lists.partition(targetClients, httpSyncProperties.getNotifyBatchSize());               // batch execution                partitionClients.forEach(item -> scheduler.execute(() -> doRun(item)));            } else {                // execute task                doRun(clients);            }        }
        private void doRun(final Collection<LongPollingClient> clients) {            // Notify all clients that a data change has occurred            for (Iterator<LongPollingClient> iter = clients.iterator(); iter.hasNext();) {                LongPollingClient client = iter.next();                iter.remove();                // send response to client                client.sendResponse(Collections.singletonList(groupKey));                Log.info("send response with the changed group,ip={}, group={}, changeTime={}", client.ip, groupKey, changeTime);            }        }    }

/** * Shenyu SpringMvc Client Configuration */@Configuration@ImportAutoConfiguration(ShenyuClientCommonBeanConfiguration.class)@ConditionalOnProperty(value = "shenyu.register.enabled", matchIfMissing = true, havingValue = "true")public class ShenyuSpringMvcClientConfiguration {
    // create SpringMvcClientEventListener to handle metadata and URI    @Bean    public SpringMvcClientEventListener springHttpClientEventListener(final ShenyuClientConfig clientConfig,                                                                      final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        return new SpringMvcClientEventListener(clientConfig.getClient().get(RpcTypeEnum.HTTP.getName()), shenyuClientRegisterRepository);    }}

/** * Shenyu Client Common Bean Configuration */@Configurationpublic class ShenyuClientCommonBeanConfiguration {       // create ShenyuClientRegisterRepository by factory     @Bean    public ShenyuClientRegisterRepository shenyuClientRegisterRepository(final ShenyuRegisterCenterConfig config) {        return ShenyuClientRegisterRepositoryFactory.newInstance(config);    }        // create ShenyuRegisterCenterConfig to read shenyu.register properties    @Bean    @ConfigurationProperties(prefix = "shenyu.register")    public ShenyuRegisterCenterConfig shenyuRegisterCenterConfig() {        return new ShenyuRegisterCenterConfig();    }      // create ShenyuClientConfig to read shenyu.client properties    @Bean    @ConfigurationProperties(prefix = "shenyu")    public ShenyuClientConfig shenyuClientConfig() {        return new ShenyuClientConfig();    }}

/** * load ShenyuClientRegisterRepository */public final class ShenyuClientRegisterRepositoryFactory {        private static final Map<String, ShenyuClientRegisterRepository> REPOSITORY_MAP = new ConcurrentHashMap<>();        /**     * create ShenyuClientRegisterRepository     */    public static ShenyuClientRegisterRepository newInstance(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig) {        if (!REPOSITORY_MAP.containsKey(shenyuRegisterCenterConfig.getRegisterType())) {            // Loading by means of SPI, type determined by registerType            ShenyuClientRegisterRepository result = ExtensionLoader.getExtensionLoader(ShenyuClientRegisterRepository.class).getJoin(shenyuRegisterCenterConfig.getRegisterType());            //init ShenyuClientRegisterRepository            result.init(shenyuRegisterCenterConfig);            ShenyuClientShutdownHook.set(result, shenyuRegisterCenterConfig.getProps());            REPOSITORY_MAP.put(shenyuRegisterCenterConfig.getRegisterType(), result);            return result;        }        return REPOSITORY_MAP.get(shenyuRegisterCenterConfig.getRegisterType());    }}

@Joinpublic class HttpClientRegisterRepository implements ShenyuClientRegisterRepository {
    @Override    public void init(final ShenyuRegisterCenterConfig config) {        this.username = config.getProps().getProperty(Constants.USER_NAME);        this.password = config.getProps().getProperty(Constants.PASS_WORD);        this.serverList = Lists.newArrayList(Splitter.on(",").split(config.getServerLists()));        this.setAccessToken();    }    // ......}

@Configuration@ImportAutoConfiguration(ShenyuClientCommonBeanConfiguration.class)public class ShenyuSpringMvcClientConfiguration {     // ......        // create SpringMvcClientEventListener    @Bean    public SpringMvcClientEventListener springHttpClientEventListener(final ShenyuClientConfig clientConfig,                                                                      final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        return new SpringMvcClientEventListener(clientConfig.getClient().get(RpcTypeEnum.HTTP.getName()), shenyuClientRegisterRepository);    }}

public abstract class AbstractContextRefreshedEventListener<T, A extends Annotation> implements ApplicationListener<ContextRefreshedEvent> {
    //......
    // Instantiation is done through the constructor    public AbstractContextRefreshedEventListener(final PropertiesConfig clientConfig,                                                 final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        // read shenyu.client.http properties        Properties props = clientConfig.getProps();        // appName         this.appName = props.getProperty(ShenyuClientConstants.APP_NAME);        // contextPath        this.contextPath = Optional.ofNullable(props.getProperty(ShenyuClientConstants.CONTEXT_PATH)).map(UriUtils::repairData).orElse("");        if (StringUtils.isBlank(appName) && StringUtils.isBlank(contextPath)) {            String errorMsg = "client register param must config the appName or contextPath";            LOG.error(errorMsg);            throw new ShenyuClientIllegalArgumentException(errorMsg);        }        this.ipAndPort = props.getProperty(ShenyuClientConstants.IP_PORT);        // host        this.host = props.getProperty(ShenyuClientConstants.HOST);        // port        this.port = props.getProperty(ShenyuClientConstants.PORT);        // publish event        publisher.start(shenyuClientRegisterRepository);    }
    // This method is executed when a context refresh event(ContextRefreshedEvent), occurs    @Override    public void onApplicationEvent(@NonNull final ContextRefreshedEvent event) {        // The contents of the method are guaranteed to be executed only once        if (!registered.compareAndSet(false, true)) {            return;        }        final ApplicationContext context = event.getApplicationContext();        // get the specific beans         Map<String, T> beans = getBeans(context);        if (MapUtils.isEmpty(beans)) {            return;        }        // build URI data and register it        publisher.publishEvent(buildURIRegisterDTO(context, beans));        // build metadata and register it        beans.forEach(this::handle);    }        @SuppressWarnings("all")    protected abstract URIRegisterDTO buildURIRegisterDTO(ApplicationContext context,                                                          Map<String, T> beans);

    protected void handle(final String beanName, final T bean) {        Class<?> clazz = getCorrectedClass(bean);        final A beanShenyuClient = AnnotatedElementUtils.findMergedAnnotation(clazz, getAnnotationType());        final String superPath = buildApiSuperPath(clazz, beanShenyuClient);        if (Objects.nonNull(beanShenyuClient) && superPath.contains("*")) {            handleClass(clazz, bean, beanShenyuClient, superPath);            return;        }        final Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(clazz);        for (Method method : methods) {            handleMethod(bean, clazz, beanShenyuClient, method, superPath);        }    }
    // default implementation. build URI data and register it    protected void handleClass(final Class<?> clazz,                               final T bean,                               @NonNull final A beanShenyuClient,                               final String superPath) {        publisher.publishEvent(buildMetaDataDTO(bean, beanShenyuClient, pathJoin(contextPath, superPath), clazz, null));    }
    // default implementation. build metadata and register it    protected void handleMethod(final T bean,                                final Class<?> clazz,                                @Nullable final A beanShenyuClient,                                final Method method,                                final String superPath) {        // get the annotation        A methodShenyuClient = AnnotatedElementUtils.findMergedAnnotation(method, getAnnotationType());        if (Objects.nonNull(methodShenyuClient)) {            // 构建元数据，发送注册事件            publisher.publishEvent(buildMetaDataDTO(bean, methodShenyuClient, buildApiPath(method, superPath, methodShenyuClient), clazz, method));        }    }        protected abstract MetaDataRegisterDTO buildMetaDataDTO(T bean,                                                            @NonNull A shenyuClient,                                                            String path,                                                            Class<?> clazz,                                                            Method method);}

public class ShenyuClientRegisterEventPublisher {        private static final ShenyuClientRegisterEventPublisher INSTANCE = new ShenyuClientRegisterEventPublisher();
    private DisruptorProviderManage<DataTypeParent> providerManage;        public static ShenyuClientRegisterEventPublisher getInstance() {        return INSTANCE;    }        public void start(final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        RegisterClientExecutorFactory factory = new RegisterClientExecutorFactory();        factory.addSubscribers(new ShenyuClientMetadataExecutorSubscriber(shenyuClientRegisterRepository));        factory.addSubscribers(new ShenyuClientURIExecutorSubscriber(shenyuClientRegisterRepository));        providerManage = new DisruptorProviderManage(factory);        providerManage.startup();    }        public <T> void publishEvent(final DataTypeParent data) {        DisruptorProvider<DataTypeParent> provider = providerManage.getProvider();        provider.onData(data);    }}

public class SpringMvcClientEventListener extends AbstractContextRefreshedEventListener<Object, ShenyuSpringMvcClient> {        private final List<Class<? extends Annotation>> mappingAnnotation = new ArrayList<>(3);        private final Boolean isFull;        private final String protocol;        // 构造函数    public SpringMvcClientEventListener(final PropertiesConfig clientConfig,                                        final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        super(clientConfig, shenyuClientRegisterRepository);        Properties props = clientConfig.getProps();        // get isFull        this.isFull = Boolean.parseBoolean(props.getProperty(ShenyuClientConstants.IS_FULL, Boolean.FALSE.toString()));        // http protocol        this.protocol = props.getProperty(ShenyuClientConstants.PROTOCOL, ShenyuClientConstants.HTTP);        mappingAnnotation.add(ShenyuSpringMvcClient.class);        mappingAnnotation.add(RequestMapping.class);    }        @Override    protected Map<String, Object> getBeans(final ApplicationContext context) {        // Configuration attribute, if isFull=true, means register the whole microservice        if (Boolean.TRUE.equals(isFull)) {            getPublisher().publishEvent(MetaDataRegisterDTO.builder()                    .contextPath(getContextPath())                    .appName(getAppName())                    .path(PathUtils.decoratorPathWithSlash(getContextPath()))                    .rpcType(RpcTypeEnum.HTTP.getName())                    .enabled(true)                    .ruleName(getContextPath())                    .build());            return null;        }        // get bean with Controller annotation        return context.getBeansWithAnnotation(Controller.class);    }        @Override    protected URIRegisterDTO buildURIRegisterDTO(final ApplicationContext context,                                                 final Map<String, Object> beans) {        // ...    }        @Override    protected String buildApiSuperPath(final Class<?> clazz, @Nullable final ShenyuSpringMvcClient beanShenyuClient) {        if (Objects.nonNull(beanShenyuClient) && StringUtils.isNotBlank(beanShenyuClient.path())) {            return beanShenyuClient.path();        }        RequestMapping requestMapping = AnnotationUtils.findAnnotation(clazz, RequestMapping.class);        // Only the first path is supported temporarily        if (Objects.nonNull(requestMapping) && ArrayUtils.isNotEmpty(requestMapping.path()) && StringUtils.isNotBlank(requestMapping.path()[0])) {            return requestMapping.path()[0];        }        return "";    }        @Override    protected Class<ShenyuSpringMvcClient> getAnnotationType() {        return ShenyuSpringMvcClient.class;    }        @Override    protected void handleMethod(final Object bean, final Class<?> clazz,                                @Nullable final ShenyuSpringMvcClient beanShenyuClient,                                final Method method, final String superPath) {        // get RequestMapping annotation        final RequestMapping requestMapping = AnnotatedElementUtils.findMergedAnnotation(method, RequestMapping.class);        // get ShenyuSpringMvcClient annotation        ShenyuSpringMvcClient methodShenyuClient = AnnotatedElementUtils.findMergedAnnotation(method, ShenyuSpringMvcClient.class);        methodShenyuClient = Objects.isNull(methodShenyuClient) ? beanShenyuClient : methodShenyuClient;        // the result of ReflectionUtils#getUniqueDeclaredMethods contains method such as hashCode, wait, toSting        // add Objects.nonNull(requestMapping) to make sure not register wrong method        if (Objects.nonNull(methodShenyuClient) && Objects.nonNull(requestMapping)) {            getPublisher().publishEvent(buildMetaDataDTO(bean, methodShenyuClient, buildApiPath(method, superPath, methodShenyuClient), clazz, method));        }    }        //...        // 构造元数据    @Override    protected MetaDataRegisterDTO buildMetaDataDTO(final Object bean,                                                   @NonNull final ShenyuSpringMvcClient shenyuClient,                                                   final String path, final Class<?> clazz,                                                   final Method method) {        //...    }}

/** * ShenyuSpringMvcClient */@Retention(RetentionPolicy.RUNTIME)@Target({ElementType.TYPE, ElementType.METHOD})public @interface ShenyuSpringMvcClient {
    // path    @AliasFor(attribute = "path")    String value() default "";        // path    @AliasFor(attribute = "value")    String path();        // ruleName    String ruleName() default "";        // desc info    String desc() default "";
    // enabled    boolean enabled() default true;        // register MetaData     boolean  registerMetaData() default false;}

@RestController@RequestMapping("/test")@ShenyuSpringMvcClient(path = "/test/**")  // register the entire interfacepublic class HttpTestController {    //......}

@RestController@RequestMapping("/order")@ShenyuSpringMvcClient(path = "/order")public class OrderController {
    /**     * Save order dto.     *     * @param orderDTO the order dto     * @return the order dto     */    @PostMapping("/save")    @ShenyuSpringMvcClient(path = "/save", desc = "Save order") // register current method    public OrderDTO save(@RequestBody final OrderDTO orderDTO) {        orderDTO.setName("hello world save order");        return orderDTO;    }

package com.lmax.disruptor;
public interface WorkHandler<T> {    void onEvent(T event) throws Exception;}

/** *  * QueueConsumer */public class QueueConsumer<T> implements WorkHandler<DataEvent<T>> {        // ......
    @Override    public void onEvent(final DataEvent<T> t) {        if (t != null) {            // Use different thread pools based on DataEvent type            ThreadPoolExecutor executor = orderly(t);            // create queue consumption tasks via factory            QueueConsumerExecutor<T> queueConsumerExecutor = factory.create();            // set data            queueConsumerExecutor.setData(t.getData());            // help gc            t.setData(null);            // put in the thread pool to execute the consumption task            executor.execute(queueConsumerExecutor);        }    }}

public final class RegisterClientConsumerExecutor<T extends DataTypeParent> extends QueueConsumerExecutor<T> {        //...... 
    @Override    public void run() {        // get data        final T data = getData();        // call the appropriate processor for processing according to the data type        subscribers.get(data.getType()).executor(Lists.newArrayList(data));    }    }

public enum DataType {       META_DATA,        URI,}

public class ShenyuClientMetadataExecutorSubscriber implements ExecutorTypeSubscriber<MetaDataRegisterDTO> {       //......        @Override    public DataType getType() {        return DataType.META_DATA;    }        @Override    public void executor(final Collection<MetaDataRegisterDTO> metaDataRegisterDTOList) {        for (MetaDataRegisterDTO metaDataRegisterDTO : metaDataRegisterDTOList) {            // call the interface method persistInterface() to finish publishing the data            shenyuClientRegisterRepository.persistInterface(metaDataRegisterDTO);        }    }}

/** * load ShenyuClientRegisterRepository */public final class ShenyuClientRegisterRepositoryFactory {        private static final Map<String, ShenyuClientRegisterRepository> REPOSITORY_MAP = new ConcurrentHashMap<>();        /**     * create ShenyuClientRegisterRepository     */    public static ShenyuClientRegisterRepository newInstance(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig) {        if (!REPOSITORY_MAP.containsKey(shenyuRegisterCenterConfig.getRegisterType())) {            // loading by means of SPI, type determined by registerType            ShenyuClientRegisterRepository result = ExtensionLoader.getExtensionLoader(ShenyuClientRegisterRepository.class).getJoin(shenyuRegisterCenterConfig.getRegisterType());            // perform initialization operations            result.init(shenyuRegisterCenterConfig);            ShenyuClientShutdownHook.set(result, shenyuRegisterCenterConfig.getProps());            REPOSITORY_MAP.put(shenyuRegisterCenterConfig.getRegisterType(), result);            return result;        }        return REPOSITORY_MAP.get(shenyuRegisterCenterConfig.getRegisterType());    }}

@Joinpublic class HttpClientRegisterRepository extends FailbackRegistryRepository {
    private static final Logger LOGGER = LoggerFactory.getLogger(HttpClientRegisterRepository.class);
    private static URIRegisterDTO uriRegisterDTO;
    private String username;
    private String password;
    private List<String> serverList;
    private String accessToken;        public HttpClientRegisterRepository() {    }        public HttpClientRegisterRepository(final ShenyuRegisterCenterConfig config) {        init(config);    }
    @Override    public void init(final ShenyuRegisterCenterConfig config) {        // admin username        this.username = config.getProps().getProperty(Constants.USER_NAME);        // admin paaword        this.password = config.getProps().getProperty(Constants.PASS_WORD);        // admin server address        this.serverList = Lists.newArrayList(Splitter.on(",").split(config.getServerLists()));        // set access token        this.setAccessToken();    }
    /**     * Persist uri.     *     * @param registerDTO the register dto     */    @Override    public void doPersistURI(final URIRegisterDTO registerDTO) {        if (RuntimeUtils.listenByOther(registerDTO.getPort())) {            return;        }        doRegister(registerDTO, Constants.URI_PATH, Constants.URI);        uriRegisterDTO = registerDTO;    }
    @Override    public void doPersistInterface(final MetaDataRegisterDTO metadata) {        doRegister(metadata, Constants.META_PATH, Constants.META_TYPE);    }
    @Override    public void close() {        if (uriRegisterDTO != null) {            uriRegisterDTO.setEventType(EventType.DELETED);            doRegister(uriRegisterDTO, Constants.URI_PATH, Constants.URI);        }    }
    private void setAccessToken() {        for (String server : serverList) {            try {                Optional<?> login = RegisterUtils.doLogin(username, password, server.concat(Constants.LOGIN_PATH));                login.ifPresent(v -> this.accessToken = String.valueOf(v));            } catch (Exception e) {                LOGGER.error("Login admin url :{} is fail, will retry. cause: {} ", server, e.getMessage());            }        }    }
    private <T> void doRegister(final T t, final String path, final String type) {        int i = 0;        // iterate through the list of admin services (admin may be clustered)        for (String server : serverList) {            i++;            String concat = server.concat(path);            try {                // 设置访问token                if (StringUtils.isBlank(accessToken)) {                    this.setAccessToken();                    if (StringUtils.isBlank(accessToken)) {                        throw new NullPointerException("accessToken is null");                    }                }                // calling the tool class to send http requests                RegisterUtils.doRegister(GsonUtils.getInstance().toJson(t), concat, type, accessToken);                return;            } catch (Exception e) {                LOGGER.error("Register admin url :{} is fail, will retry. cause:{}", server, e.getMessage());                if (i == serverList.size()) {                    throw new RuntimeException(e);                }            }        }    }}

public final class RegisterUtils {      //...... 
    // Sending data via OkHttp    public static void doRegister(final String json, final String url, final String type) throws IOException {        if (!StringUtils.hasLength(accessToken)) {            LOGGER.error("{} client register error accessToken is null, please check the config : {} ", type, json);            return;        }        Headers headers = new Headers.Builder().add(Constants.X_ACCESS_TOKEN, accessToken).build();        String result = OkHttpTools.getInstance().post(url, json, headers);        if (Objects.equals(SUCCESS, result)) {            LOGGER.info("{} client register success: {} ", type, json);        } else {            LOGGER.error("{} client register error: {} ", type, json);        }    }}

public class ShenyuClientURIExecutorSubscriber implements ExecutorTypeSubscriber<URIRegisterDTO> {        //......        @Override    public DataType getType() {        return DataType.URI;     }        // register URI    @Override    public void executor(final Collection<URIRegisterDTO> dataList) {        for (URIRegisterDTO uriRegisterDTO : dataList) {            Stopwatch stopwatch = Stopwatch.createStarted();            while (true) {                try (Socket ignored = new Socket(uriRegisterDTO.getHost(), uriRegisterDTO.getPort())) {                    break;                } catch (IOException e) {                    long sleepTime = 1000;                    // maybe the port is delay exposed                    if (stopwatch.elapsed(TimeUnit.SECONDS) > 5) {                        LOG.error("host:{}, port:{} connection failed, will retry",                                uriRegisterDTO.getHost(), uriRegisterDTO.getPort());                        // If the connection fails for a long time, Increase sleep time                        if (stopwatch.elapsed(TimeUnit.SECONDS) > 180) {                            sleepTime = 10000;                        }                    }                    try {                        TimeUnit.MILLISECONDS.sleep(sleepTime);                    } catch (InterruptedException ex) {                        ex.printStackTrace();                    }                }            }            ShenyuClientShutdownHook.delayOtherHooks();                        shenyuClientRegisterRepository.persistURI(uriRegisterDTO);        }    }}

@RequestMapping("/shenyu-client")@Joinpublic class ShenyuHttpRegistryController implements ShenyuServerRegisterRepository {
    private ShenyuServerRegisterPublisher publisher;
    @Override    public void init(final ShenyuServerRegisterPublisher publisher, final ShenyuRegisterCenterConfig config) {        this.publisher = publisher;    }        // register Metadata    @PostMapping("/register-metadata")    @ResponseBody    public String registerMetadata(@RequestBody final MetaDataRegisterDTO metaDataRegisterDTO) {        publisher.publish(metaDataRegisterDTO);        return ShenyuResultMessage.SUCCESS;    }           // register URI    @PostMapping("/register-uri")    @ResponseBody    public String registerURI(@RequestBody final URIRegisterDTO uriRegisterDTO) {        publisher.publish(uriRegisterDTO);        return ShenyuResultMessage.SUCCESS;    }}

@Configurationpublic class RegisterCenterConfiguration {    @Bean    @ConfigurationProperties(prefix = "shenyu.register")    public ShenyuRegisterCenterConfig shenyuRegisterCenterConfig() {        return new ShenyuRegisterCenterConfig();    }        //create ShenyuServerRegisterRepository to register in admin    @Bean(destroyMethod = "close")    public ShenyuServerRegisterRepository shenyuServerRegisterRepository(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig, final List<ShenyuClientRegisterService> shenyuClientRegisterService) {        // 1. get the registration type from the configuration property        String registerType = shenyuRegisterCenterConfig.getRegisterType();        // 2. load the implementation class by registering the type with the SPI method        ShenyuServerRegisterRepository registerRepository = ExtensionLoader.getExtensionLoader(ShenyuServerRegisterRepository.class).getJoin(registerType);        // 3. get the publisher and write data to the Disruptor queue        RegisterServerDisruptorPublisher publisher = RegisterServerDisruptorPublisher.getInstance();        // 4. ShenyuClientRegisterService, rpcType -> registerService        Map<String, ShenyuClientRegisterService> registerServiceMap = shenyuClientRegisterService.stream().collect(Collectors.toMap(ShenyuClientRegisterService::rpcType, e -> e));        // 5. start publisher        publisher.start(registerServiceMap);        // 6. init registerRepository        registerRepository.init(publisher, shenyuRegisterCenterConfig);        return registerRepository;    }}

public class RegisterClientServerDisruptorPublisher implements ShenyuServerRegisterPublisher {    private static final RegisterClientServerDisruptorPublisher INSTANCE = new     private static final RegisterClientServerDisruptorPublisher INSTANCE = new RegisterServerDisruptorPublisher();();
    public static RegisterClientServerDisruptorPublisher getInstance() {        return INSTANCE;    }       //prepare for event publishing, add server-side metadata and URI subscribers, process data. And start the Disruptor queue.    public void start(final Map<String, ShenyuClientRegisterService> shenyuClientRegisterService) {        RegisterServerExecutorFactory factory = new RegisterServerExecutorFactory();        // add URI data subscriber        factory.addSubscribers(new URIRegisterExecutorSubscriber(shenyuClientRegisterService));        // add Metadata subscriber        factory.addSubscribers(new MetadataExecutorSubscriber(shenyuClientRegisterService));        //start Disruptor        providerManage = new DisruptorProviderManage(factory);        providerManage.startup();    }        // write data to queue    @Override    public <T> void publish(final DataTypeParent data) {        DisruptorProvider<Object> provider = providerManage.getProvider();        provider.onData(Collections.singleton(data));    }
    // write data to queue on batch    @Override    public void publish(final Collection<? extends DataTypeParent> dataList) {        DisruptorProvider<Collection<DataTypeParent>> provider = providerManage.getProvider();        provider.onData(dataList.stream().map(DataTypeParent.class::cast).collect(Collectors.toList()));    }        @Override    public void close() {        providerManage.getProvider().shutdown();    }}

package com.lmax.disruptor;
public interface WorkHandler<T> {    void onEvent(T var1) throws Exception;}

/** *  * QueueConsumer */public class QueueConsumer<T> implements WorkHandler<DataEvent<T>> {        // ......
    @Override    public void onEvent(final DataEvent<T> t) {        if (t != null) {            // Use different thread pools based on DataEvent type            ThreadPoolExecutor executor = orderly(t);            // create queue consumption tasks via factory            QueueConsumerExecutor<T> queueConsumerExecutor = factory.create();            // set data            queueConsumerExecutor.setData(t.getData());            // help gc            t.setData(null);            // put in the thread pool to execute the consumption task            executor.execute(queueConsumerExecutor);        }    }}

public final class RegisterServerConsumerExecutor extends QueueConsumerExecutor<List<DataTypeParent>> {   // ...
    @Override    public void run() {        //get the data from the disruptor queue and check data        Collection<DataTypeParent> results = getData()                .stream()                .filter(this::isValidData)                .collect(Collectors.toList());        if (CollectionUtils.isEmpty(results)) {            return;        }        //execute operations according to type        getType(results).executor(results);    }        // get subscribers by type    private ExecutorSubscriber<DataTypeParent> selectExecutor(final Collection<DataTypeParent> list) {        final Optional<DataTypeParent> first = list.stream().findFirst();        return subscribers.get(first.orElseThrow(() -> new RuntimeException("the data type is not found")).getType());    }}

public class MetadataExecutorSubscriber implements ExecutorTypeSubscriber<MetaDataRegisterDTO> {     //......
    @Override    public DataType getType() {        return DataType.META_DATA;     }
    @Override    public void executor(final Collection<MetaDataRegisterDTO> metaDataRegisterDTOList) {        // Traversing the metadata list        metaDataRegisterDTOList.forEach(meta -> {            Optional.ofNullable(this.shenyuClientRegisterService.get(meta.getRpcType())) // Get registered Service by type                    .ifPresent(shenyuClientRegisterService -> {                        // Registration of metadata, locking to ensure sequential execution and prevent concurrent errors                        synchronized (shenyuClientRegisterService) {                            shenyuClientRegisterService.register(meta);                        }                    });        });    }}

public class URIRegisterExecutorSubscriber implements ExecutorTypeSubscriber<URIRegisterDTO> {    //......        @Override    public DataType getType() {        return DataType.URI;     }        @Override    public void executor(final Collection<URIRegisterDTO> dataList) {        if (CollectionUtils.isEmpty(dataList)) {            return;        }                findService(dataList).ifPresent(service -> {            Map<String, List<URIRegisterDTO>> listMap = buildData(dataList);            listMap.forEach(service::registerURI);        });        final Map<String, List<URIRegisterDTO>> groupByRpcType = dataList.stream()                .filter(data -> StringUtils.isNotBlank(data.getRpcType()))                .collect(Collectors.groupingBy(URIRegisterDTO::getRpcType));        for (Map.Entry<String, List<URIRegisterDTO>> entry : groupByRpcType.entrySet()) {            final String rpcType = entry.getKey();            // Get registered Service by type            Optional.ofNullable(shenyuClientRegisterService.get(rpcType))                    .ifPresent(service -> {                        final List<URIRegisterDTO> list = entry.getValue();                        // Build URI data types and register them with the registerURI method                        Map<String, List<URIRegisterDTO>> listMap = buildData(list);                        listMap.forEach(service::registerURI);                    });        }    }        // Find Service by type    private Optional<ShenyuClientRegisterService> findService(final Collection<URIRegisterDTO> dataList) {        return dataList.stream().map(dto -> shenyuClientRegisterService.get(dto.getRpcType())).findFirst();    }}

public abstract class AbstractShenyuClientRegisterServiceImpl extends FallbackShenyuClientRegisterService implements ShenyuClientRegisterService {
    //......
    public String register(final MetaDataRegisterDTO dto) {        // 1.register selector information        String selectorHandler = selectorHandler(dto);        String selectorId = selectorService.registerDefault(dto, PluginNameAdapter.rpcTypeAdapter(rpcType()), selectorHandler);        // 2.register rule information        String ruleHandler = ruleHandler();        RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);        ruleService.registerDefault(ruleDTO);        // 3.register metadata information        registerMetadata(dto);        // 4.register contextPath        String contextPath = dto.getContextPath();        if (StringUtils.isNotEmpty(contextPath)) {            registerContextPath(dto);        }        return ShenyuResultMessage.SUCCESS;    }}

public abstract class AbstractShenyuClientRegisterServiceImpl extends FallbackShenyuClientRegisterService implements ShenyuClientRegisterService {
    //......
    public String registerURI(final String selectorName, final List<URIRegisterDTO> uriList) {        if (CollectionUtils.isEmpty(uriList)) {            return "";        }        // Does the corresponding selector exist        SelectorDO selectorDO = selectorService.findByNameAndPluginName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));        if (Objects.isNull(selectorDO)) {            return "";        }        // Handle handler information in the selector        String handler = buildHandle(uriList, selectorDO);        selectorDO.setHandle(handler);        SelectorData selectorData = selectorService.buildByName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));        selectorData.setHandle(handler);
        // Update records in the database        selectorService.updateSelective(selectorDO);        // publish Event to gateway        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE, Collections.singletonList(selectorData)));        return ShenyuResultMessage.SUCCESS;    }}