/** * DubboTestServiceImpl. */@Service("dubboTestService")public class DubboTestServiceImpl implements DubboTestService {        @Override    @ShenyuDubboClient(path = "/findById", desc = "Query by Id")    public DubboTest findById(final String id) {        return new DubboTest(id, "hello world shenyu Apache, findById");    }
    //......}

// dubbo服务@Service("dubboTestService")public class DubboTestServiceImpl implements DubboTestService {        @Override    @ShenyuDubboClient(path = "/findById", desc = "Query by Id") // 需要注册的方法    public DubboTest findById(final String id) {        return new DubboTest(id, "hello world shenyu Apache, findById");    }
    //......}

/** * 作用于类和方法上 */@Retention(RetentionPolicy.RUNTIME)@Target({ElementType.TYPE, ElementType.METHOD})@Inheritedpublic @interface ShenyuDubboClient {        //注册路径    String path();        //规则名称    String ruleName() default "";       //描述信息    String desc() default "";
    //是否启用    boolean enabled() default true;}

public class ApacheDubboServiceBeanListener implements ApplicationListener<ContextRefreshedEvent> {
    // ......
    //构造器    public ApacheDubboServiceBeanListener(final PropertiesConfig clientConfig, final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        //1.读取属性配置        Properties props = clientConfig.getProps();        String contextPath = props.getProperty(ShenyuClientConstants.CONTEXT_PATH);        String appName = props.getProperty(ShenyuClientConstants.APP_NAME);        if (StringUtils.isBlank(contextPath)) {            throw new ShenyuClientIllegalArgumentException("apache dubbo client must config the contextPath or appName");        }        this.contextPath = contextPath;        this.appName = appName;        this.host = props.getProperty(ShenyuClientConstants.HOST);        this.port = props.getProperty(ShenyuClientConstants.PORT);        //2.开启线程池        executorService = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("shenyu-apache-dubbo-client-thread-pool-%d").build());        //3.启动注册中心        publisher.start(shenyuClientRegisterRepository);    }
    /**     * 上下文刷新事件，执行方法逻辑     */    @Override    public void onApplicationEvent(final ContextRefreshedEvent contextRefreshedEvent) {        //......    }

    @Override    public void onApplicationEvent(final ContextRefreshedEvent contextRefreshedEvent) {        //读取ServiceBean        Map<String, ServiceBean> serviceBean = contextRefreshedEvent.getApplicationContext().getBeansOfType(ServiceBean.class);        if (serviceBean.isEmpty()) {            return;        }        //保证该方法只执行一次        if (!registered.compareAndSet(false, true)) {            return;        }        //处理元数据对象        for (Map.Entry<String, ServiceBean> entry : serviceBean.entrySet()) {            handler(entry.getValue());        }        //处理URI对象        serviceBean.values().stream().findFirst().ifPresent(bean -> {            publisher.publishEvent(buildURIRegisterDTO(bean));        });    }

    private void handler(final ServiceBean<?> serviceBean) {        //获取代理对象        Object refProxy = serviceBean.getRef();        //获取class信息        Class<?> clazz = refProxy.getClass();        if (AopUtils.isAopProxy(refProxy)) {            clazz = AopUtils.getTargetClass(refProxy);        }        //获取所有方法        Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(clazz);        for (Method method : methods) {            //读取ShenyuDubboClient注解信息            ShenyuDubboClient shenyuDubboClient = method.getAnnotation(ShenyuDubboClient.class);            if (Objects.nonNull(shenyuDubboClient)) {                //构建元数据对象，并注册                publisher.publishEvent(buildMetaDataDTO(serviceBean, shenyuDubboClient, method));            }        }    }

    private MetaDataRegisterDTO buildMetaDataDTO(final ServiceBean<?> serviceBean, final ShenyuDubboClient shenyuDubboClient, final Method method) {        //应用名称        String appName = buildAppName(serviceBean);        //方法路径        String path = contextPath + shenyuDubboClient.path();        //描述信息        String desc = shenyuDubboClient.desc();        //服务名称        String serviceName = serviceBean.getInterface();        //规则名称        String configRuleName = shenyuDubboClient.ruleName();        String ruleName = ("".equals(configRuleName)) ? path : configRuleName;        //方法名称        String methodName = method.getName();        //参数类型        Class<?>[] parameterTypesClazz = method.getParameterTypes();        String parameterTypes = Arrays.stream(parameterTypesClazz).map(Class::getName).collect(Collectors.joining(","));        return MetaDataRegisterDTO.builder()                .appName(appName)                .serviceName(serviceName)                .methodName(methodName)                .contextPath(contextPath)                .host(buildHost())                .port(buildPort(serviceBean))                .path(path)                .ruleName(ruleName)                .pathDesc(desc)                .parameterTypes(parameterTypes)                .rpcExt(buildRpcExt(serviceBean)) //dubbo服务的扩展信息                .rpcType(RpcTypeEnum.DUBBO.getName())                .enabled(shenyuDubboClient.enabled())                .build();    }

private URIRegisterDTO buildURIRegisterDTO(final ServiceBean serviceBean) {        return URIRegisterDTO.builder()                .contextPath(this.contextPath) //上下文路径                .appName(buildAppName(serviceBean))//应用名称                .rpcType(RpcTypeEnum.DUBBO.getName())//rpc类型：dubbo                .host(buildHost()) //host                .port(buildPort(serviceBean))//port                .build(); }

//向注册中心，发布注册事件   publisher.publishEvent();

   @Override    public String register(final MetaDataRegisterDTO dto) {        //1. 注册选择器        String selectorHandler = selectorHandler(dto);        String selectorId = selectorService.registerDefault(dto, PluginNameAdapter.rpcTypeAdapter(rpcType()), selectorHandler);        //2. 注册规则        String ruleHandler = ruleHandler();        RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);        ruleService.registerDefault(ruleDTO);        //3. 注册元数据        registerMetadata(dto);        //4. 注册ContextPath        String contextPath = dto.getContextPath();        if (StringUtils.isNotEmpty(contextPath)) {            registerContextPath(dto);        }        return ShenyuResultMessage.SUCCESS;    }

    @Override    public String registerDefault(final MetaDataRegisterDTO dto, final String pluginName, final String selectorHandler) {        // 构建contextPath        String contextPath = ContextPathUtils.buildContextPath(dto.getContextPath(), dto.getAppName());        // 通过名称查找选择器信息是否存在        SelectorDO selectorDO = findByNameAndPluginName(contextPath, pluginName);        if (Objects.isNull(selectorDO)) {            // 不存在就创建默认的选择器信息            return registerSelector(contextPath, pluginName, selectorHandler);        }        return selectorDO.getId();    }

   //注册选择器   private String registerSelector(final String contextPath, final String pluginName, final String selectorHandler) {        //构建选择器        SelectorDTO selectorDTO = buildSelectorDTO(contextPath, pluginMapper.selectByName(pluginName).getId());        selectorDTO.setHandle(selectorHandler);        //注册默认选择器        return registerDefault(selectorDTO);    }     //构建选择器    private SelectorDTO buildSelectorDTO(final String contextPath, final String pluginId) {        //构建默认选择器        SelectorDTO selectorDTO = buildDefaultSelectorDTO(contextPath);        selectorDTO.setPluginId(pluginId);         //构建默认选择器的条件属性        selectorDTO.setSelectorConditions(buildDefaultSelectorConditionDTO(contextPath));        return selectorDTO;    }

private SelectorDTO buildDefaultSelectorDTO(final String name) {    return SelectorDTO.builder()            .name(name) // 名称            .type(SelectorTypeEnum.CUSTOM_FLOW.getCode()) // 默认类型自定义            .matchMode(MatchModeEnum.AND.getCode()) //默认匹配方式 and            .enabled(Boolean.TRUE)  //默认启开启            .loged(Boolean.TRUE)  //默认记录日志            .continued(Boolean.TRUE) //默认继续后续选择器            .sort(1) //默认顺序1            .build();}

private List<SelectorConditionDTO> buildDefaultSelectorConditionDTO(final String contextPath) {    SelectorConditionDTO selectorConditionDTO = new SelectorConditionDTO();    selectorConditionDTO.setParamType(ParamTypeEnum.URI.getName()); // 默认参数类型URI    selectorConditionDTO.setParamName("/");    selectorConditionDTO.setOperator(OperatorEnum.MATCH.getAlias()); // 默认匹配策略 match    selectorConditionDTO.setParamValue(contextPath + AdminConstants.URI_SUFFIX); // 默认值 /contextPath/**    return Collections.singletonList(selectorConditionDTO);}

@Overridepublic String registerDefault(final SelectorDTO selectorDTO) {    //选择器信息    SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);    //选择器条件属性    List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();    if (StringUtils.isEmpty(selectorDTO.getId())) {        // 向数据库插入选择器信息        selectorMapper.insertSelective(selectorDO);          // 向数据库插入选择器条件属性        selectorConditionDTOs.forEach(selectorConditionDTO -> {            selectorConditionDTO.setSelectorId(selectorDO.getId());            selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));        });    }    // 发布同步事件，向网关同步选择信息及其条件属性    publishEvent(selectorDO, selectorConditionDTOs);    return selectorDO.getId();}

@Override    public String register(final MetaDataRegisterDTO dto) {        //1. 注册选择器        //......                //2. 注册规则        // 默认规则处理属性        String ruleHandler = ruleHandler();        // 构建默认规则信息        RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);        // 注册规则        ruleService.registerDefault(ruleDTO);                //3. 注册元数据        //......                //4. 注册ContextPath        //......                return ShenyuResultMessage.SUCCESS;    }

    @Override    protected String ruleHandler() {        // 默认规则处理属性        return new DubboRuleHandle().toJson();    }

public class DubboRuleHandle implements RuleHandle {
    /**     * dubbo服务版本信息.     */    private String version;
    /**     * 分组.     */    private String group;
    /**     * 重试次数.     */    private Integer retries = 0;
    /**     * 负载均衡策略：默认随机     */    private String loadbalance = LoadBalanceEnum.RANDOM.getName();
    /**     * 超时，默认3000     */    private long timeout = Constants.TIME_OUT;}

  // 构建默认规则信息    private RuleDTO buildRpcDefaultRuleDTO(final String selectorId, final MetaDataRegisterDTO metaDataDTO, final String ruleHandler) {        return buildRuleDTO(selectorId, ruleHandler, metaDataDTO.getRuleName(), metaDataDTO.getPath());    }   //  构建默认规则信息    private RuleDTO buildRuleDTO(final String selectorId, final String ruleHandler, final String ruleName, final String path) {        RuleDTO ruleDTO = RuleDTO.builder()                .selectorId(selectorId) //关联的选择器id                .name(ruleName) //规则名称                .matchMode(MatchModeEnum.AND.getCode()) // 默认匹配模式 and                .enabled(Boolean.TRUE) // 默认开启                .loged(Boolean.TRUE) //默认记录日志                .sort(1) //默认顺序 1                .handle(ruleHandler)                .build();        RuleConditionDTO ruleConditionDTO = RuleConditionDTO.builder()                .paramType(ParamTypeEnum.URI.getName()) // 默认参数类型URI                .paramName("/")                .paramValue(path) //参数值path                .build();        if (path.indexOf("*") > 1) {            ruleConditionDTO.setOperator(OperatorEnum.MATCH.getAlias()); //如果path中有*，操作类型则默认为 match        } else {            ruleConditionDTO.setOperator(OperatorEnum.EQ.getAlias()); // 否则，默认操作类型 =         }        ruleDTO.setRuleConditions(Collections.singletonList(ruleConditionDTO));        return ruleDTO;    }

    @Override    public String registerDefault(final RuleDTO ruleDTO) {        RuleDO exist = ruleMapper.findBySelectorIdAndName(ruleDTO.getSelectorId(), ruleDTO.getName());        if (Objects.nonNull(exist)) {            return "";        }
        RuleDO ruleDO = RuleDO.buildRuleDO(ruleDTO);        List<RuleConditionDTO> ruleConditions = ruleDTO.getRuleConditions();        if (StringUtils.isEmpty(ruleDTO.getId())) {            // 向数据库插入规则信息            ruleMapper.insertSelective(ruleDO);            //向数据库插入规则体条件属性            ruleConditions.forEach(ruleConditionDTO -> {                ruleConditionDTO.setRuleId(ruleDO.getId());                ruleConditionMapper.insertSelective(RuleConditionDO.buildRuleConditionDO(ruleConditionDTO));            });        }        // 向网关发布事件，进行数据同步        publishEvent(ruleDO, ruleConditions);        return ruleDO.getId();    }

   @Override    public String register(final MetaDataRegisterDTO dto) {        //1. 注册选择器        //......                //2. 注册规则        //......                //3. 注册元数据        registerMetadata(dto);                //4. 注册ContextPath        //......                return ShenyuResultMessage.SUCCESS;    }

    @Override    protected void registerMetadata(final MetaDataRegisterDTO dto) {            // 获取metaDataService            MetaDataService metaDataService = getMetaDataService();            // 元数据是否存在            MetaDataDO exist = metaDataService.findByPath(dto.getPath());            // 插入或更新元数据            metaDataService.saveOrUpdateMetaData(exist, dto);    }
    @Override    public void saveOrUpdateMetaData(final MetaDataDO exist, final MetaDataRegisterDTO metaDataDTO) {        DataEventTypeEnum eventType;        // 数据类型转换 DTO->DO        MetaDataDO metaDataDO = MetaDataTransfer.INSTANCE.mapRegisterDTOToEntity(metaDataDTO);        // 插入数据        if (Objects.isNull(exist)) {            Timestamp currentTime = new Timestamp(System.currentTimeMillis());            metaDataDO.setId(UUIDUtils.getInstance().generateShortUuid());            metaDataDO.setDateCreated(currentTime);            metaDataDO.setDateUpdated(currentTime);            metaDataMapper.insert(metaDataDO);            eventType = DataEventTypeEnum.CREATE;        } else {            // 更新数据            metaDataDO.setId(exist.getId());            metaDataMapper.update(metaDataDO);            eventType = DataEventTypeEnum.UPDATE;        }        // 发布同步事件到网关        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.META_DATA, eventType,                Collections.singletonList(MetaDataTransfer.INSTANCE.mapToData(metaDataDO))));    }

    @Override    public String registerURI(final String selectorName, final List<URIRegisterDTO> uriList) {        String result;        String key = key(selectorName);        try {            this.removeFallBack(key);            // 注册URI            result = this.doRegisterURI(selectorName, uriList);            logger.info("Register success: {},{}", selectorName, uriList);        } catch (Exception ex) {            logger.warn("Register exception: cause:{}", ex.getMessage());            result = "";            // 注册失败后，进行重试            this.addFallback(key, new FallbackHolder(selectorName, uriList));        }        return result;    }

@Override    public String doRegisterURI(final String selectorName, final List<URIRegisterDTO> uriList) {        //参数检查        if (CollectionUtils.isEmpty(uriList)) {            return "";        }        //获取选择器信息        SelectorDO selectorDO = selectorService.findByNameAndPluginName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));        if (Objects.isNull(selectorDO)) {            throw new ShenyuException("doRegister Failed to execute,wait to retry.");        }        // 获取有效的URI        List<URIRegisterDTO> validUriList = uriList.stream().filter(dto -> Objects.nonNull(dto.getPort()) && StringUtils.isNotBlank(dto.getHost())).collect(Collectors.toList());        // 构建选择器的handle属性        String handler = buildHandle(validUriList, selectorDO);        if (handler != null) {            selectorDO.setHandle(handler);            SelectorData selectorData = selectorService.buildByName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));            selectorData.setHandle(handler);            // 向数据库更新选择器的handle属性            selectorService.updateSelective(selectorDO);            // 向网关发送选择器更新事件            eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE, Collections.singletonList(selectorData)));        }        return ShenyuResultMessage.SUCCESS;    }

public final class ShenyuWebHandler implements WebHandler, ApplicationListener<SortPluginEvent> {    //......        /**     * 处理web请求     */    @Override    public Mono<Void> handle(@NonNull final ServerWebExchange exchange) {       // 执行默认插件链        Mono<Void> execute = new DefaultShenyuPluginChain(plugins).execute(exchange);        if (scheduled) {            return execute.subscribeOn(scheduler);        }        return execute;    }        private static class DefaultShenyuPluginChain implements ShenyuPluginChain {
        private int index;
        private final List<ShenyuPlugin> plugins;
        /**         * 实例化默认插件链         */        DefaultShenyuPluginChain(final List<ShenyuPlugin> plugins) {            this.plugins = plugins;        }
        /**         * 执行每个插件.         */        @Override        public Mono<Void> execute(final ServerWebExchange exchange) {            return Mono.defer(() -> {                if (this.index < plugins.size()) {                    // 获取当前执行插件                    ShenyuPlugin plugin = plugins.get(this.index++);                    // 是否跳过当前插件                    boolean skip = plugin.skip(exchange);                    if (skip) {                        // 如果跳过就执行下一个                        return this.execute(exchange);                    }                    // 执行当前插件                    return plugin.execute(exchange, this);                }                return Mono.empty();            });        }    }}

@Override    public Mono<Void> execute(final ServerWebExchange exchange, final ShenyuPluginChain chain) {        // 插件名称        String pluginName = named();        // 插件信息        PluginData pluginData = BaseDataCache.getInstance().obtainPluginData(pluginName);        if (pluginData != null && pluginData.getEnabled()) {            // 选择器信息            final Collection<SelectorData> selectors = BaseDataCache.getInstance().obtainSelectorData(pluginName);            if (CollectionUtils.isEmpty(selectors)) {                return handleSelectorIfNull(pluginName, exchange, chain);            }            // 匹配选择器            SelectorData selectorData = matchSelector(exchange, selectors);            if (Objects.isNull(selectorData)) {                return handleSelectorIfNull(pluginName, exchange, chain);            }            selectorLog(selectorData, pluginName);            // 规则信息            List<RuleData> rules = BaseDataCache.getInstance().obtainRuleData(selectorData.getId());            if (CollectionUtils.isEmpty(rules)) {                return handleRuleIfNull(pluginName, exchange, chain);            }            // 匹配规则            RuleData rule;            if (selectorData.getType() == SelectorTypeEnum.FULL_FLOW.getCode()) {                //get last                rule = rules.get(rules.size() - 1);            } else {                rule = matchRule(exchange, rules);            }            if (Objects.isNull(rule)) {                return handleRuleIfNull(pluginName, exchange, chain);            }            ruleLog(rule, pluginName);            // 执行插件            return doExecute(exchange, chain, selectorData, rule);        }        return chain.execute(exchange);    }

public class GlobalPlugin implements ShenyuPlugin {    // 构建上下文信息    private final ShenyuContextBuilder builder;        //......        @Override    public Mono<Void> execute(final ServerWebExchange exchange, final ShenyuPluginChain chain) {       // 构建上下文信息，传入到 exchange 中        ShenyuContext shenyuContext = builder.build(exchange);        exchange.getAttributes().put(Constants.CONTEXT, shenyuContext);        return chain.execute(exchange);    }        //......}

public class DefaultShenyuContextBuilder implements ShenyuContextBuilder {    //......        @Override    public ShenyuContext build(final ServerWebExchange exchange) {        //构建参数        Pair<String, MetaData> buildData = buildData(exchange);        //包装ShenyuContext        return decoratorMap.get(buildData.getLeft()).decorator(buildDefaultContext(exchange.getRequest()), buildData.getRight());    }        private Pair<String, MetaData> buildData(final ServerWebExchange exchange) {        //......        //根据请求的uri获取元数据        MetaData metaData = MetaDataCache.getInstance().obtain(request.getURI().getPath());        if (Objects.nonNull(metaData) && Boolean.TRUE.equals(metaData.getEnabled())) {            exchange.getAttributes().put(Constants.META_DATA, metaData);            return Pair.of(metaData.getRpcType(), metaData);        } else {            return Pair.of(RpcTypeEnum.HTTP.getName(), new MetaData());        }    }    //设置默认的上下文信息    private ShenyuContext buildDefaultContext(final ServerHttpRequest request) {        String appKey = request.getHeaders().getFirst(Constants.APP_KEY);        String sign = request.getHeaders().getFirst(Constants.SIGN);        String timestamp = request.getHeaders().getFirst(Constants.TIMESTAMP);        ShenyuContext shenyuContext = new ShenyuContext();        String path = request.getURI().getPath();        shenyuContext.setPath(path); //请求路径        shenyuContext.setAppKey(appKey);        shenyuContext.setSign(sign);        shenyuContext.setTimestamp(timestamp);        shenyuContext.setStartDateTime(LocalDateTime.now());        Optional.ofNullable(request.getMethod()).ifPresent(httpMethod -> shenyuContext.setHttpMethod(httpMethod.name()));//请求方法        return shenyuContext;    } }

public class DubboShenyuContextDecorator implements ShenyuContextDecorator {        @Override    public ShenyuContext decorator(final ShenyuContext shenyuContext, final MetaData metaData) {        shenyuContext.setModule(metaData.getAppName());//获取AppName        shenyuContext.setMethod(metaData.getServiceName()); //获取ServiceName        shenyuContext.setContextPath(metaData.getContextPath()); //获取contextPath        shenyuContext.setRpcType(RpcTypeEnum.DUBBO.getName()); // dubbo服务        return shenyuContext;    }        @Override    public String rpcType() {        return RpcTypeEnum.DUBBO.getName();    }}

public class RpcParamTransformPlugin implements ShenyuPlugin {
    @Override    public Mono<Void> execute(final ServerWebExchange exchange, final ShenyuPluginChain chain) {        //从exchange中获取请求信息        ServerHttpRequest request = exchange.getRequest();        ShenyuContext shenyuContext = exchange.getAttribute(Constants.CONTEXT);        if (Objects.nonNull(shenyuContext)) {           // 如果请求参数格式是APPLICATION_JSON            MediaType mediaType = request.getHeaders().getContentType();            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {                return body(exchange, request, chain);            }            // 如果请求参数格式是APPLICATION_FORM_URLENCODED            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {                return formData(exchange, request, chain);            }            //一般查询请求            return query(exchange, request, chain);        }        return chain.execute(exchange);    }        // 如果请求参数格式是APPLICATION_JSON    private Mono<Void> body(final ServerWebExchange exchange, final ServerHttpRequest serverHttpRequest, final ShenyuPluginChain chain) {        return Mono.from(DataBufferUtils.join(serverHttpRequest.getBody())                .flatMap(body -> {                    exchange.getAttributes().put(Constants.PARAM_TRANSFORM, resolveBodyFromRequest(body));//解析body，保存到exchange中                    return chain.execute(exchange);                }));    }   // 如果请求参数格式是APPLICATION_FORM_URLENCODED    private Mono<Void> formData(final ServerWebExchange exchange, final ServerHttpRequest serverHttpRequest, final ShenyuPluginChain chain) {        return Mono.from(DataBufferUtils.join(serverHttpRequest.getBody())                .flatMap(map -> {                    String param = resolveBodyFromRequest(map);                    LinkedMultiValueMap<String, String> linkedMultiValueMap;                    try {                        linkedMultiValueMap = BodyParamUtils.buildBodyParams(URLDecoder.decode(param, StandardCharsets.UTF_8.name())); //格式化数据                    } catch (UnsupportedEncodingException e) {                        return Mono.error(e);                    }                    exchange.getAttributes().put(Constants.PARAM_TRANSFORM, HttpParamConverter.toMap(() -> linkedMultiValueMap));// 保存到exchange中                    return chain.execute(exchange);                }));    }    //一般查询请求    private Mono<Void> query(final ServerWebExchange exchange, final ServerHttpRequest serverHttpRequest, final ShenyuPluginChain chain) {        exchange.getAttributes().put(Constants.PARAM_TRANSFORM, HttpParamConverter.ofString(() -> serverHttpRequest.getURI().getQuery()));//保存到exchange中        return chain.execute(exchange);    }    //...... }

public abstract class AbstractDubboPlugin extends AbstractShenyuPlugin {        @Override    public Mono<Void> doExecute(final ServerWebExchange exchange,                                   final ShenyuPluginChain chain,                                   final SelectorData selector,                                   final RuleData rule) {        //获取参数        String param = exchange.getAttribute(Constants.PARAM_TRANSFORM);        //获取上下文信息        ShenyuContext shenyuContext = exchange.getAttribute(Constants.CONTEXT);        assert shenyuContext != null;        //获取元数据        MetaData metaData = exchange.getAttribute(Constants.META_DATA);        //检查元数据        if (!checkMetaData(metaData)) {            LOG.error(" path is : {}, meta data have error : {}", shenyuContext.getPath(), metaData);            exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);            Object error = ShenyuResultWrap.error(exchange, ShenyuResultEnum.META_DATA_ERROR, null);            return WebFluxResultUtils.result(exchange, error);        }        //检查元数据和参数        if (Objects.nonNull(metaData) && StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(param)) {            exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);            Object error = ShenyuResultWrap.error(exchange, ShenyuResultEnum.DUBBO_HAVE_BODY_PARAM, null);            return WebFluxResultUtils.result(exchange, error);        }        //设置rpcContext        this.rpcContext(exchange);        //进行dubbo服务调用        return this.doDubboInvoker(exchange, chain, selector, rule, metaData, param);    }}

public class ApacheDubboPlugin extends AbstractDubboPlugin {        @Override    protected Mono<Void> doDubboInvoker(final ServerWebExchange exchange,                                        final ShenyuPluginChain chain,                                        final SelectorData selector,                                        final RuleData rule,                                        final MetaData metaData,                                        final String param) {        //设置当前的选择器和规则信息，以及请求地址，用于支持dubbo的灰度        RpcContext.getContext().setAttachment(Constants.DUBBO_SELECTOR_ID, selector.getId());        RpcContext.getContext().setAttachment(Constants.DUBBO_RULE_ID, rule.getId());        RpcContext.getContext().setAttachment(Constants.DUBBO_REMOTE_ADDRESS, Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress());        //dubbo的泛化调用        final Mono<Object> result = dubboProxyService.genericInvoker(param, metaData, exchange);        //执行下一个插件        return result.then(chain.execute(exchange));    }}

public class ApacheDubboProxyService {    //...... 
    /**     * Generic invoker object.     */    public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws ShenyuException {        //1.获取ReferenceConfig对象        ReferenceConfig<GenericService> reference = ApacheDubboConfigCache.getInstance().get(metaData.getPath());        //如果没有获取到        if (Objects.isNull(reference) || StringUtils.isEmpty(reference.getInterface())) {            //失效当前缓存的信息            ApacheDubboConfigCache.getInstance().invalidate(metaData.getPath());            //使用元数据进行再次初始化            reference = ApacheDubboConfigCache.getInstance().initRef(metaData);        }        //2.获取泛化服务GenericService对象        GenericService genericService = reference.get();        //3.构造请求参数pair对象        Pair<String[], Object[]> pair;        if (StringUtils.isBlank(metaData.getParameterTypes()) || ParamCheckUtils.dubboBodyIsEmpty(body)) {            pair = new ImmutablePair<>(new String[]{}, new Object[]{});        } else {            pair = dubboParamResolveService.buildParameter(body, metaData.getParameterTypes());        }        //4.发起异步的泛化调用        return Mono.fromFuture(invokeAsync(genericService, metaData.getMethodName(), pair.getLeft(), pair.getRight()).thenApply(ret -> {            //处理结果            if (Objects.isNull(ret)) {                ret = Constants.DUBBO_RPC_RESULT_EMPTY;            }            exchange.getAttributes().put(Constants.RPC_RESULT, ret);            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());            return ret;        })).onErrorMap(exception -> exception instanceof GenericException ? new ShenyuException(((GenericException) exception).getExceptionMessage()) : new ShenyuException(exception));//处理异常    }        //泛化调用，异步操作    private CompletableFuture<Object> invokeAsync(final GenericService genericService, final String method, final String[] parameterTypes, final Object[] args) throws GenericException {        genericService.$invoke(method, parameterTypes, args);        Object resultFromFuture = RpcContext.getContext().getFuture();        return resultFromFuture instanceof CompletableFuture ? (CompletableFuture<Object>) resultFromFuture : CompletableFuture.completedFuture(resultFromFuture);    }}

public abstract class AbstractDubboPluginDataHandler implements PluginDataHandler {    //......        //初始化配置缓存   protected abstract void initConfigCache(DubboRegisterConfig dubboRegisterConfig);
    @Override    public void handlerPlugin(final PluginData pluginData) {        if (Objects.nonNull(pluginData) && Boolean.TRUE.equals(pluginData.getEnabled())) {            //数据反序列化            DubboRegisterConfig dubboRegisterConfig = GsonUtils.getInstance().fromJson(pluginData.getConfig(), DubboRegisterConfig.class);            DubboRegisterConfig exist = Singleton.INST.get(DubboRegisterConfig.class);            if (Objects.isNull(dubboRegisterConfig)) {                return;            }            if (Objects.isNull(exist) || !dubboRegisterConfig.equals(exist)) {                // 执行初始化操作                this.initConfigCache(dubboRegisterConfig);            }            Singleton.INST.single(DubboRegisterConfig.class, dubboRegisterConfig);        }    }    //......}

public class ApacheDubboPluginDataHandler extends AbstractDubboPluginDataHandler {
    @Override    protected void initConfigCache(final DubboRegisterConfig dubboRegisterConfig) {        //执行初始化操作        ApacheDubboConfigCache.getInstance().init(dubboRegisterConfig);        //失效之前缓存的结果        ApacheDubboConfigCache.getInstance().invalidateAll();    }}

public final class ApacheDubboConfigCache extends DubboConfigCache {    //......     /**     * 初始化     */    public void init(final DubboRegisterConfig dubboRegisterConfig) {        //创建ApplicationConfig        if (Objects.isNull(applicationConfig)) {            applicationConfig = new ApplicationConfig("shenyu_proxy");        }        //协议或者地址发生改变时，需要更新registryConfig        if (needUpdateRegistryConfig(dubboRegisterConfig)) {            RegistryConfig registryConfigTemp = new RegistryConfig();            registryConfigTemp.setProtocol(dubboRegisterConfig.getProtocol());            registryConfigTemp.setId("shenyu_proxy");            registryConfigTemp.setRegister(false);            registryConfigTemp.setAddress(dubboRegisterConfig.getRegister());            Optional.ofNullable(dubboRegisterConfig.getGroup()).ifPresent(registryConfigTemp::setGroup);            registryConfig = registryConfigTemp;        }        //创建ConsumerConfig        if (Objects.isNull(consumerConfig)) {            consumerConfig = ApplicationModel.getConfigManager().getDefaultConsumer().orElseGet(() -> {                ConsumerConfig consumerConfig = new ConsumerConfig();                consumerConfig.refresh();                return consumerConfig;            });            //设置ConsumerConfig            Optional.ofNullable(dubboRegisterConfig.getThreadpool()).ifPresent(consumerConfig::setThreadpool);             Optional.ofNullable(dubboRegisterConfig.getCorethreads()).ifPresent(consumerConfig::setCorethreads);             Optional.ofNullable(dubboRegisterConfig.getThreads()).ifPresent(consumerConfig::setThreads);             Optional.ofNullable(dubboRegisterConfig.getQueues()).ifPresent(consumerConfig::setQueues);        }    }        //是否需要更新注册配置    private boolean needUpdateRegistryConfig(final DubboRegisterConfig dubboRegisterConfig) {        if (Objects.isNull(registryConfig)) {            return true;        }        return !Objects.equals(dubboRegisterConfig.getProtocol(), registryConfig.getProtocol())                || !Objects.equals(dubboRegisterConfig.getRegister(), registryConfig.getAddress())                || !Objects.equals(dubboRegisterConfig.getProtocol(), registryConfig.getProtocol());    }
    //......}

public class ApacheDubboMetaDataSubscriber implements MetaDataSubscriber {    //本地内存缓存    private static final ConcurrentMap<String, MetaData> META_DATA = Maps.newConcurrentMap();
    //元数据发生更新    public void onSubscribe(final MetaData metaData) {        // dubbo服务的元数据更新        if (RpcTypeEnum.DUBBO.getName().equals(metaData.getRpcType())) {            //对应的元数据是否存在            MetaData exist = META_DATA.get(metaData.getPath());            if (Objects.isNull(exist) || Objects.isNull(ApacheDubboConfigCache.getInstance().get(metaData.getPath()))) {                // 首次初始化                ApacheDubboConfigCache.getInstance().initRef(metaData);            } else {                // 对应的元数据发生了更新操作                if (!Objects.equals(metaData.getServiceName(), exist.getServiceName())                        || !Objects.equals(metaData.getRpcExt(), exist.getRpcExt())                        || !Objects.equals(metaData.getParameterTypes(), exist.getParameterTypes())                        || !Objects.equals(metaData.getMethodName(), exist.getMethodName())) {                    //根据最新的元数据再次构建ReferenceConfig                    ApacheDubboConfigCache.getInstance().build(metaData);                }            }            //本地内存缓存            META_DATA.put(metaData.getPath(), metaData);        }    }
    //删除元数据    public void unSubscribe(final MetaData metaData) {        if (RpcTypeEnum.DUBBO.getName().equals(metaData.getRpcType())) {            //使ReferenceConfig失效            ApacheDubboConfigCache.getInstance().invalidate(metaData.getPath());            META_DATA.remove(metaData.getPath());        }    }}

public final class ApacheDubboConfigCache extends DubboConfigCache {    //......        public ReferenceConfig<GenericService> initRef(final MetaData metaData) {            try {                //先尝试从缓存中获取，存在就直接返回                ReferenceConfig<GenericService> referenceConfig = cache.get(metaData.getPath());                if (StringUtils.isNoneBlank(referenceConfig.getInterface())) {                    return referenceConfig;                }            } catch (ExecutionException e) {                LOG.error("init dubbo ref exception", e);            }           //不存在，就构建            return build(metaData);        }
        /**         * Build reference config.         */        @SuppressWarnings("deprecation")        public ReferenceConfig<GenericService> build(final MetaData metaData) {            if (Objects.isNull(applicationConfig) || Objects.isNull(registryConfig)) {                return new ReferenceConfig<>();            }            ReferenceConfig<GenericService> reference = new ReferenceConfig<>(); //新建ReferenceConfig            reference.setGeneric("true"); //泛化调用            reference.setAsync(true);//支持异步
            reference.setApplication(applicationConfig);//设置应用配置            reference.setRegistry(registryConfig);//设置注册中心配置            reference.setConsumer(consumerConfig);//设置消费者配置            reference.setInterface(metaData.getServiceName());//设置服务接口            reference.setProtocol("dubbo");//设置dubbo协议            reference.setCheck(false); //不检查 service provider            reference.setLoadbalance("gray");//支持灰度
            Map<String, String> parameters = new HashMap<>(2);            parameters.put("dispatcher", "direct");            reference.setParameters(parameters);//自定义参数
            String rpcExt = metaData.getRpcExt();//rpc扩展参数            DubboParam dubboParam = parserToDubboParam(rpcExt);//反序列化            if (Objects.nonNull(dubboParam)) {                if (StringUtils.isNoneBlank(dubboParam.getVersion())) {                    reference.setVersion(dubboParam.getVersion());//设置版本                }                if (StringUtils.isNoneBlank(dubboParam.getGroup())) {                    reference.setGroup(dubboParam.getGroup());//设置分组                }                if (StringUtils.isNoneBlank(dubboParam.getUrl())) {                    reference.setUrl(dubboParam.getUrl());//设置url                }                if (StringUtils.isNoneBlank(dubboParam.getCluster())) {                    reference.setCluster(dubboParam.getCluster());//设置Cluster type                }                Optional.ofNullable(dubboParam.getTimeout()).ifPresent(reference::setTimeout);//timeout                Optional.ofNullable(dubboParam.getRetries()).ifPresent(reference::setRetries);//retires                Optional.ofNullable(dubboParam.getSent()).ifPresent(reference::setSent);//Whether to ack async-sent            }            try {                //获取GenericService                Object obj = reference.get();                if (Objects.nonNull(obj)) {                    LOG.info("init apache dubbo reference success there meteData is :{}", metaData);                    //缓存当前的reference                    cache.put(metaData.getPath(), reference);                }            } catch (Exception e) {                LOG.error("init apache dubbo reference exception", e);            }            return reference;        }    //......    }

    @Override    public Mono<Void> execute(final ServerWebExchange exchange, final ShenyuPluginChain chain) {        ShenyuContext shenyuContext = exchange.getAttribute(Constants.CONTEXT);        assert shenyuContext != null;        // 根据rpc类型处理结果        return writerMap.get(shenyuContext.getRpcType()).writeWith(exchange, chain);    }

public class RPCMessageWriter implements MessageWriter {
    @Override    public Mono<Void> writeWith(final ServerWebExchange exchange, final ShenyuPluginChain chain) {        return chain.execute(exchange).then(Mono.defer(() -> {            Object result = exchange.getAttribute(Constants.RPC_RESULT); //获取结果            if (Objects.isNull(result)) { //处理异常                Object error = ShenyuResultWrap.error(exchange, ShenyuResultEnum.SERVICE_RESULT_ERROR, null);                return WebFluxResultUtils.result(exchange, error);            }            return WebFluxResultUtils.result(exchange, result);//返回结果        }));    }}

/** * shenyu 客户端http注册配置类 */@Configuration@ImportAutoConfiguration(ShenyuClientCommonBeanConfiguration.class)@ConditionalOnProperty(value = "shenyu.register.enabled", matchIfMissing = true, havingValue = "true")public class ShenyuSpringMvcClientConfiguration {       // 创建SpringMvcClientEventListener，主要处理元数据和URI信息   @Bean   public SpringMvcClientEventListener springHttpClientEventListener(final ShenyuClientConfig clientConfig,                                                                     final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {       return new SpringMvcClientEventListener(clientConfig.getClient().get(RpcTypeEnum.HTTP.getName()), shenyuClientRegisterRepository);   }}

/** * shenyu客户端通用配置类 */@Configurationpublic class ShenyuClientCommonBeanConfiguration {       // 创建ShenyuClientRegisterRepository，通过工厂类创建而成。    @Bean    public ShenyuClientRegisterRepository shenyuClientRegisterRepository(final ShenyuRegisterCenterConfig config) {        return ShenyuClientRegisterRepositoryFactory.newInstance(config);    }        // 创建ShenyuRegisterCenterConfig，读取shenyu.register属性配置    @Bean    @ConfigurationProperties(prefix = "shenyu.register")    public ShenyuRegisterCenterConfig shenyuRegisterCenterConfig() {        return new ShenyuRegisterCenterConfig();    }      // 创建ShenyuClientConfig，读取shenyu.client属性配置    @Bean    @ConfigurationProperties(prefix = "shenyu")    public ShenyuClientConfig shenyuClientConfig() {        return new ShenyuClientConfig();    }}

/** * 加载 ShenyuClientRegisterRepository */public final class ShenyuClientRegisterRepositoryFactory {        private static final Map<String, ShenyuClientRegisterRepository> REPOSITORY_MAP = new ConcurrentHashMap<>();        /**     * 创建 ShenyuClientRegisterRepository     */    public static ShenyuClientRegisterRepository newInstance(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig) {        if (!REPOSITORY_MAP.containsKey(shenyuRegisterCenterConfig.getRegisterType())) {            // 通过SPI的方式进行加载，类型由registerType决定            ShenyuClientRegisterRepository result = ExtensionLoader.getExtensionLoader(ShenyuClientRegisterRepository.class).getJoin(shenyuRegisterCenterConfig.getRegisterType());            //执行初始化操作            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)@ConditionalOnProperty(value = "shenyu.register.enabled", matchIfMissing = true, havingValue = "true")public class ShenyuSpringMvcClientConfiguration {     // ......        //  创建 SpringMvcClientEventListener    @Bean    public SpringMvcClientEventListener springHttpClientEventListener(final ShenyuClientConfig clientConfig,                                                                      final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        return new SpringMvcClientEventListener(clientConfig.getClient().get(RpcTypeEnum.HTTP.getName()), shenyuClientRegisterRepository);    }}

// 实现了ApplicationListener接口public abstract class AbstractContextRefreshedEventListener<T, A extends Annotation> implements ApplicationListener<ContextRefreshedEvent> {
     //......
    //构造函数    public AbstractContextRefreshedEventListener(final PropertiesConfig clientConfig,                                                 final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        // 读取 shenyu.client.http 配置信息        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);        // 开始事件发布        publisher.start(shenyuClientRegisterRepository);    }
    // 当有上下文刷新事件ContextRefreshedEvent发生时，该方法会执行    @Override    public void onApplicationEvent(@NonNull final ContextRefreshedEvent event) {        //保证该方法的内容只执行一次        if (!registered.compareAndSet(false, true)) {            return;        }        final ApplicationContext context = event.getApplicationContext();        // 获取声明RPC调用的类        Map<String, T> beans = getBeans(context);        if (MapUtils.isEmpty(beans)) {            return;        }        // 构建URI数据并注册        publisher.publishEvent(buildURIRegisterDTO(context, beans));        // 构建元数据并注册        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);        // 获取当前bean的对应shenyu客户端的注解（对应不同的RPC调用注解不一样，像http的就是@ShenyuSpringMvcClient,而像SpringCloud的则是@ShenyuSpringCloudClient）        final A beanShenyuClient = AnnotatedElementUtils.findMergedAnnotation(clazz, getAnnotationType());        // 根据bean获取对应的path（不同子类实现不一样）        final String superPath = buildApiSuperPath(clazz, beanShenyuClient);        // 如果包含Shenyu客户端注解或者path中包括'*'，表示注册整个类的接口        if (Objects.nonNull(beanShenyuClient) && superPath.contains("*")) {            // 构建类的元数据，发送注册事件            handleClass(clazz, bean, beanShenyuClient, superPath);            return;        }        // 获取当前bean的所有方法        final Method[] methods = ReflectionUtils.getUniqueDeclaredMethods(clazz);        // 遍历方法        for (Method method : methods) {            // 注册符合条件的方法            handleMethod(bean, clazz, beanShenyuClient, method, superPath);        }    }
    // 构建类元数据并注册的默认实现    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));    }
    // 构建方法元数据并注册的默认实现    protected void handleMethod(final T bean,                                final Class<?> clazz,                                @Nullable final A beanShenyuClient,                                final Method method,                                final String superPath) {        // 如果方法上有Shenyu客户端注解，就表示该方法需要注册        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;        /**     * 公开静态方法     *     * @return ShenyuClientRegisterEventPublisher instance     */    public static ShenyuClientRegisterEventPublisher getInstance() {        return INSTANCE;    }        /**     * Start方法执行     *     * @param shenyuClientRegisterRepository shenyuClientRegisterRepository     */    public void start(final ShenyuClientRegisterRepository shenyuClientRegisterRepository) {        // 创建客户端注册工厂类        RegisterClientExecutorFactory factory = new RegisterClientExecutorFactory();        // 添加元数据订阅器        factory.addSubscribers(new ShenyuClientMetadataExecutorSubscriber(shenyuClientRegisterRepository));        //  添加URI订阅器        factory.addSubscribers(new ShenyuClientURIExecutorSubscriber(shenyuClientRegisterRepository));        // 启动Disruptor队列        providerManage = new DisruptorProviderManage(factory);        providerManage.startup();    }        /**     * 发布事件，向Disruptor队列发数据     *     * @param data the data     */    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();        // 获取 isFull        this.isFull = Boolean.parseBoolean(props.getProperty(ShenyuClientConstants.IS_FULL, Boolean.FALSE.toString()));        // 表示是http协议的实现        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) {        // 配置属性，如果 isFull=true 的话，表示注册整个微服务        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;        }        // 否则获取带Controller注解的bean        return context.getBeansWithAnnotation(Controller.class);    }        // 构造URI数据    @Override    protected URIRegisterDTO buildURIRegisterDTO(final ApplicationContext context,                                                 final Map<String, Object> beans) {        // ...    }        @Override    protected String buildApiSuperPath(final Class<?> clazz, @Nullable final ShenyuSpringMvcClient beanShenyuClient) {        // 如果有带上Shenyu客户端注解，则优先取注解中的不为空的path属性        if (Objects.nonNull(beanShenyuClient) && StringUtils.isNotBlank(beanShenyuClient.path())) {            return beanShenyuClient.path();        }        // 如果有带上RequestMapping注解，且path属性不为空，则返回path数组的第一个值        RequestMapping requestMapping = AnnotationUtils.findAnnotation(clazz, RequestMapping.class);        if (Objects.nonNull(requestMapping) && ArrayUtils.isNotEmpty(requestMapping.path()) && StringUtils.isNotBlank(requestMapping.path()[0])) {            return requestMapping.path()[0];        }        return "";    }        // 声明http实现的客户端注解是ShenyuSpringMvcClient    @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) {        // 获取当前bean的RequestMapping注解        final RequestMapping requestMapping = AnnotatedElementUtils.findMergedAnnotation(method, RequestMapping.class);        // 获取当前bean的 ShenyuSpringMvcClient 注解        ShenyuSpringMvcClient methodShenyuClient = AnnotatedElementUtils.findMergedAnnotation(method, ShenyuSpringMvcClient.class);        methodShenyuClient = Objects.isNull(methodShenyuClient) ? beanShenyuClient : methodShenyuClient;        //如果有 ShenyuSpringMvcClient 注解并且包含RequestMapping注解（表示是一个接口），则进行注册        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) {        //...    }}

/** * shenyu 客户端接口，用于方法上或类上 */@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 描述信息    String desc() default "";
    // enabled是否启用    boolean enabled() default true;        // registerMetaData 注册元数据    boolean  registerMetaData() default false;}

@RestController@RequestMapping("/test")@ShenyuSpringMvcClient(path = "/test/**")  // 表示整个接口注册public 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") // 注册当前方法    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;}

/** *  * 队列消费者 */public class QueueConsumer<T> implements WorkHandler<DataEvent<T>> {        // 省略了其他逻辑
    @Override    public void onEvent(final DataEvent<T> t) {        if (t != null) {            // 根据事件类型使用不同的线程池            ThreadPoolExecutor executor = orderly(t);            // 通过工厂创建队列消费任务            QueueConsumerExecutor<T> queueConsumerExecutor = factory.create();            // 保存数据            queueConsumerExecutor.setData(t.getData());            // help gc            t.setData(null);            // 放在线程池中执行 消费任务            executor.execute(queueConsumerExecutor);        }    }}

public final class RegisterClientConsumerExecutor<T extends DataTypeParent> extends QueueConsumerExecutor<T> {        //...... 
    @Override    public void run() {        // 获取数据        final T data = getData();        // 根据数据类型调用相应的处理器进行处理        subscribers.get(data.getType()).executor(Lists.newArrayList(data));    }    }

//数据类型public enum DataType {   // 元数据    META_DATA,       // URI数据    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) {            // 调用接口方法persistInterface()完成数据的发布            shenyuClientRegisterRepository.persistInterface(metaDataRegisterDTO);        }    }}

/** * 加载 ShenyuClientRegisterRepository */public final class ShenyuClientRegisterRepositoryFactory {        private static final Map<String, ShenyuClientRegisterRepository> REPOSITORY_MAP = new ConcurrentHashMap<>();        /**     * 创建 ShenyuClientRegisterRepository     */    public static ShenyuClientRegisterRepository newInstance(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig) {        if (!REPOSITORY_MAP.containsKey(shenyuRegisterCenterConfig.getRegisterType())) {            // 通过SPI的方式进行加载，类型由registerType决定            ShenyuClientRegisterRepository result = ExtensionLoader.getExtensionLoader(ShenyuClientRegisterRepository.class).getJoin(shenyuRegisterCenterConfig.getRegisterType());            //执行初始化操作            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的用户名        this.username = config.getProps().getProperty(Constants.USER_NAME);        // admin的用户名对应的密码        this.password = config.getProps().getProperty(Constants.PASS_WORD);        // admin服务列表        this.serverList = Lists.newArrayList(Splitter.on(",").split(config.getServerLists()));        // 设置访问的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;        // 遍历admin服务列表（admin可能是集群）        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");                    }                }                // 调用工具类发送 http 请求                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 {      //...... 
    // 通过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; //数据类型是URI    }        // 注册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();                    }                }            }            //添加hook，优雅停止客户端             ShenyuClientShutdownHook.delayOtherHooks();                        // 注册URI            shenyuClientRegisterRepository.persistURI(uriRegisterDTO);        }    }}

// shenuyu客户端接口@RequestMapping("/shenyu-client")@Joinpublic class ShenyuClientHttpRegistryController implements ShenyuClientServerRegisterRepository {
    private ShenyuClientServerRegisterPublisher publisher;
    @Override    public void init(final ShenyuClientServerRegisterPublisher publisher, final ShenyuRegisterCenterConfig config) {        this.publisher = publisher;    }
    @Override    public void close() {        publisher.close();    }        // 注册元数据    @PostMapping("/register-metadata")    @ResponseBody    public String registerMetadata(@RequestBody final MetaDataRegisterDTO metaDataRegisterDTO) {        publisher.publish(metaDataRegisterDTO);        return ShenyuResultMessage.SUCCESS;    }           // 注册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();    }        //创建ShenyuServerRegisterRepository，用于服务端注册    @Bean(destroyMethod = "close")    public ShenyuServerRegisterRepository shenyuServerRegisterRepository(final ShenyuRegisterCenterConfig shenyuRegisterCenterConfig, final List<ShenyuClientRegisterService> shenyuClientRegisterService) {        // 1.从配置属性中获取注册类型        String registerType = shenyuRegisterCenterConfig.getRegisterType();        // 2.通过注册类型，以SPI的方法加载实现类        ShenyuClientServerRegisterRepository registerRepository = ExtensionLoader.getExtensionLoader(ShenyuClientServerRegisterRepository.class).getJoin(registerType);        // 3.获取publisher，向Disruptor队列中写数据        RegisterClientServerDisruptorPublisher publisher = RegisterClientServerDisruptorPublisher.getInstance();        // 4.注册Service， rpcType -> registerService        Map<String, ShenyuClientRegisterService> registerServiceMap = shenyuClientRegisterService.stream().collect(Collectors.toMap(ShenyuClientRegisterService::rpcType, e -> e));        // 5.事件发布的准备工作        publisher.start(registerServiceMap);        // 6.注册的初始化操作        registerRepository.init(publisher, shenyuRegisterCenterConfig);        return registerRepository;    }}

public class RegisterClientServerDisruptorPublisher implements ShenyuClientServerRegisterPublisher {    //私有属性    private static final RegisterClientServerDisruptorPublisher INSTANCE = new RegisterClientServerDisruptorPublisher();
    private DisruptorProviderManage<Collection<DataTypeParent>> providerManage;
    //公开静态方法获取实例    public static RegisterServerDisruptorPublisher getInstance() {        return INSTANCE;    }       //事件发布的准备工作，添加服务端元数据和URI订阅器，处理数据。并且启动Disruptor队列。    public void start(final Map<String, ShenyuClientRegisterService> shenyuClientRegisterService) {        //服务端注册工厂        RegisterServerExecutorFactory factory = new RegisterServerExecutorFactory();        //添加URI数据订阅器        factory.addSubscribers(new URIRegisterExecutorSubscriber(shenyuClientRegisterService));        //添加元数据订阅器        factory.addSubscribers(new MetadataExecutorSubscriber(shenyuClientRegisterService));        //启动Disruptor队列        providerManage = new DisruptorProviderManage(factory);        providerManage.startup();    }        // 向队列中写入数据    @Override    public void publish(final DataTypeParent data) {        DisruptorProvider<Collection<DataTypeParent>> provider = providerManage.getProvider();        provider.onData(Collections.singleton(data));    }
    // 批量向队列中写入数据    @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 event) throws Exception;}

/** *  * 队列消费者 */public class QueueConsumer<T> implements WorkHandler<DataEvent<T>> {        // 省略了其他逻辑
    @Override    public void onEvent(final DataEvent<T> t) {        if (t != null) {            // 根据事件类型获取相应的线程池            ThreadPoolExecutor executor = orderly(t);            // 通过工厂创建队列消费任务            QueueConsumerExecutor<T> queueConsumerExecutor = factory.create();            // 保存数据            queueConsumerExecutor.setData(t.getData());            // help gc            t.setData(null);            // 放在线程池中执行 消费任务            executor.execute(queueConsumerExecutor);        }    }}

public final class RegisterServerConsumerExecutor extends QueueConsumerExecutor<Collection<DataTypeParent>> {   // ...
    @Override    public void run() {        //获取从disruptor队列中拿到的数据        Collection<DataTypeParent> results = getData()                .stream()                .filter(this::isValidData)                .collect(Collectors.toList());        if (CollectionUtils.isEmpty(results)) {            return;        }        //根据类型执行操作        selectExecutor(results).executor(results);    }
    // 根据类型获取订阅者    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) {        // 遍历元数据列表        metaDataRegisterDTOList.forEach(meta -> {            Optional.ofNullable(this.shenyuClientRegisterService.get(meta.getRpcType())) // 根据类型获取注册Service                    .ifPresent(shenyuClientRegisterService -> {                        // 对元数据进行注册，加锁确保顺序执行，防止并发错误                        synchronized (shenyuClientRegisterService) {                            shenyuClientRegisterService.register(meta);                        }                    });        });    }}

public class URIRegisterExecutorSubscriber implements ExecutorTypeSubscriber<URIRegisterDTO> {    //......        @Override    public DataType getType() {        return DataType.URI; // URI数据类型    }        @Override    public void executor(final Collection<URIRegisterDTO> dataList) {        if (CollectionUtils.isEmpty(dataList)) {            return;        }        // 根据rpc调用类型聚集数据        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();            // 根据类型查找Service            Optional.ofNullable(shenyuClientRegisterService.get(rpcType))                    .ifPresent(service -> {                        final List<URIRegisterDTO> list = entry.getValue();                        // 构建URI数据类型，通过registerURI方法实现注册                        Map<String, List<URIRegisterDTO>> listMap = buildData(list);                        listMap.forEach(service::registerURI);                    });        }    }}

public abstract class AbstractShenyuClientRegisterServiceImpl extends FallbackShenyuClientRegisterService implements ShenyuClientRegisterService {        //......        public String register(final MetaDataRegisterDTO dto) {        // 1.注册选择器信息        String selectorHandler = selectorHandler(dto);        String selectorId = selectorService.registerDefault(dto, PluginNameAdapter.rpcTypeAdapter(rpcType()), selectorHandler);        // 2.注册规则信息        String ruleHandler = ruleHandler();        RuleDTO ruleDTO = buildRpcDefaultRuleDTO(selectorId, dto, ruleHandler);        ruleService.registerDefault(ruleDTO);        // 3.注册元数据信息        registerMetadata(dto);        // 4.注册contextPath        String contextPath = dto.getContextPath();        if (StringUtils.isNotEmpty(contextPath)) {            registerContextPath(dto);        }        return ShenyuResultMessage.SUCCESS;    }}

public abstract class AbstractShenyuClientRegisterServiceImpl extends FallbackShenyuClientRegisterService implements ShenyuClientRegisterService {
    //......        public String doRegisterURI(final String selectorName, final List<URIRegisterDTO> uriList) {        if (CollectionUtils.isEmpty(uriList)) {            return "";        }        // 对应的选择器是否存在        SelectorDO selectorDO = selectorService.findByNameAndPluginName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));        if (Objects.isNull(selectorDO)) {            throw new ShenyuException("doRegister Failed to execute,wait to retry.");        }        List<URIRegisterDTO> validUriList = uriList.stream().filter(dto -> Objects.nonNull(dto.getPort()) && StringUtils.isNotBlank(dto.getHost())).collect(Collectors.toList());        // 处理选择器中的handler信息        String handler = buildHandle(validUriList, selectorDO);        if (handler != null) {            selectorDO.setHandle(handler);            SelectorData selectorData = selectorService.buildByName(selectorName, PluginNameAdapter.rpcTypeAdapter(rpcType()));            selectorData.setHandle(handler);            // 更新数据库中的记录            selectorService.updateSelective(selectorDO);            // 发布事件            eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE, Collections.singletonList(selectorData)));        }        return ShenyuResultMessage.SUCCESS;    }}

@Validated@RequiredArgsConstructor@RestController@RequestMapping("/selector")public class SelectorController {        @PutMapping("/{id}")    public ShenyuAdminResult updateSelector(@PathVariable("id") final String id, @Valid @RequestBody final SelectorDTO selectorDTO) {        // 设置当前选择器数据id        selectorDTO.setId(id);        // 创建或更新操作        Integer updateCount = selectorService.createOrUpdate(selectorDTO);        // 返回结果信息        return ShenyuAdminResult.success(ShenyuResultMessage.UPDATE_SUCCESS, updateCount);    }        // ......}

@RequiredArgsConstructor@Servicepublic class SelectorServiceImpl implements SelectorService {    // 负责事件发布的eventPublisher    private final ApplicationEventPublisher eventPublisher;        @Override    @Transactional(rollbackFor = Exception.class)    public int createOrUpdate(final SelectorDTO selectorDTO) {        int selectorCount;        // 构建数据 DTO --> DO        SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);        List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();        // 判断是添加还是更新        if (StringUtils.isEmpty(selectorDTO.getId())) {            // 插入选择器数据            selectorCount = selectorMapper.insertSelective(selectorDO);            // 插入选择器中的条件数据            selectorConditionDTOs.forEach(selectorConditionDTO -> {                selectorConditionDTO.setSelectorId(selectorDO.getId());                selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));            });            // check selector add            // 权限检查            if (dataPermissionMapper.listByUserId(JwtUtils.getUserInfo().getUserId()).size() > 0) {                DataPermissionDTO dataPermissionDTO = new DataPermissionDTO();                dataPermissionDTO.setUserId(JwtUtils.getUserInfo().getUserId());                dataPermissionDTO.setDataId(selectorDO.getId());                dataPermissionDTO.setDataType(AdminConstants.SELECTOR_DATA_TYPE);                dataPermissionMapper.insertSelective(DataPermissionDO.buildPermissionDO(dataPermissionDTO));            }
        } else {            // 更新数据，先删除再新增            selectorCount = selectorMapper.updateSelective(selectorDO);            //delete rule condition then add            selectorConditionMapper.deleteByQuery(new SelectorConditionQuery(selectorDO.getId()));            selectorConditionDTOs.forEach(selectorConditionDTO -> {                selectorConditionDTO.setSelectorId(selectorDO.getId());                SelectorConditionDO selectorConditionDO = SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO);                selectorConditionMapper.insertSelective(selectorConditionDO);            });        }        // 发布事件        publishEvent(selectorDO, selectorConditionDTOs);
        // 更新upstream        updateDivideUpstream(selectorDO);        return selectorCount;    }            // ......    }

     private void publishEvent(final SelectorDO selectorDO, final List<SelectorConditionDTO> selectorConditionDTOs) {        // 找到选择器对应的插件        PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());        // 构建条件数据        List<ConditionData> conditionDataList =                selectorConditionDTOs.stream().map(ConditionTransfer.INSTANCE::mapToSelectorDTO).collect(Collectors.toList());        // 发布变更数据        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,                Collections.singletonList(SelectorDO.transFrom(selectorDO, pluginDO.getName(), conditionDataList))));    }

public class DataChangedEvent extends ApplicationEvent {//......}

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
    //......    }

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
  /**     * 有数据变更时，调用此方法     * @param event     */    @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // 遍历数据变更监听器(一般使用一种数据同步的方式就好了)        for (DataChangedListener listener : listeners) {            // 哪种数据发生变更            switch (event.getGroupKey()) {                case APP_AUTH: // 认证信息                    listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());                    break;                case PLUGIN:  // 插件信息                    listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());                    break;                case RULE:    // 规则信息                    listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());                    break;                case SELECTOR:   // 选择器信息                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());                    break;                case META_DATA:  // 元数据                    listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());                    break;                default:  // 其他类型，抛出异常                    throw new IllegalStateException("Unexpected value: " + event.getGroupKey());            }        }    }    }

/** * 数据同步配置类 * 通过springboot条件装配实现 * The type Data sync configuration. */@Configurationpublic class DataSyncConfiguration {
   //省略了其他代码......      /**     * The type Nacos listener.     */    @Configuration    @ConditionalOnProperty(prefix = "shenyu.sync.nacos", name = "url")    @Import(NacosConfiguration.class)    static class NacosListener {
        /**         * Data changed listener data changed listener.         *         * @param configService the config service         * @return the data changed listener         */        @Bean        @ConditionalOnMissingBean(NacosDataChangedListener.class)        public DataChangedListener nacosDataChangedListener(final ConfigService configService) {            return new NacosDataChangedListener(configService);        }
        /**         * Nacos data init zookeeper data init.         *         * @param configService the config service         * @param syncDataService the sync data service         * @return the nacos data init         */        @Bean        @ConditionalOnMissingBean(NacosDataInit.class)        public NacosDataInit nacosDataInit(final ConfigService configService, final SyncDataService syncDataService) {            return new NacosDataInit(configService, syncDataService);        }    }         //省略了其他代码......}

/** * Nacos configuration. */@EnableConfigurationProperties(NacosProperties.class)public class NacosConfiguration {
    /**     * register configService in spring ioc.     *     * @param nacosProp the nacos configuration     * @return ConfigService {@linkplain ConfigService}     * @throws Exception the exception     */    @Bean    @ConditionalOnMissingBean(ConfigService.class)    public ConfigService nacosConfigService(final NacosProperties nacosProp) throws Exception {        Properties properties = new Properties();        if (nacosProp.getAcm() != null && nacosProp.getAcm().isEnabled()) {            // Use aliyun ACM service            properties.put(PropertyKeyConst.ENDPOINT, nacosProp.getAcm().getEndpoint());            properties.put(PropertyKeyConst.NAMESPACE, nacosProp.getAcm().getNamespace());            // Use subaccount ACM administrative authority            properties.put(PropertyKeyConst.ACCESS_KEY, nacosProp.getAcm().getAccessKey());            properties.put(PropertyKeyConst.SECRET_KEY, nacosProp.getAcm().getSecretKey());        } else {            properties.put(PropertyKeyConst.SERVER_ADDR, nacosProp.getUrl());            if (StringUtils.isNotBlank(nacosProp.getNamespace())) {                properties.put(PropertyKeyConst.NAMESPACE, nacosProp.getNamespace());            }            if (StringUtils.isNotBlank(nacosProp.getUsername())) {                properties.put(PropertyKeyConst.USERNAME, nacosProp.getUsername());            }            if (StringUtils.isNotBlank(nacosProp.getPassword())) {                properties.put(PropertyKeyConst.PASSWORD, nacosProp.getPassword());            }        }        return NacosFactory.createConfigService(properties);    }}

/** * The type Nacos config. */@ConfigurationProperties(prefix = "shenyu.sync.nacos")public class NacosProperties {
    private String url;
    private String namespace;
    private String username;
    private String password;
    private NacosACMProperties acm;
    /**     * Gets the value of url.     *     * @return the value of url     */    public String getUrl() {        return url;    }
    /**     * Sets the url.     *     * @param url url     */    public void setUrl(final String url) {        this.url = url;    }
    /**     * Gets the value of namespace.     *     * @return the value of namespace     */    public String getNamespace() {        return namespace;    }
    /**     * Sets the namespace.     *     * @param namespace namespace     */    public void setNamespace(final String namespace) {        this.namespace = namespace;    }
    /**     * Gets the value of username.     *     * @return the value of username     */    public String getUsername() {        return username;    }
    /**     * Sets the username.     *     * @param username username     */    public void setUsername(final String username) {        this.username = username;    }
    /**     * Gets the value of password.     *     * @return the value of password     */    public String getPassword() {        return password;    }
    /**     * Sets the password.     *     * @param password password     */    public void setPassword(final String password) {        this.password = password;    }
    /**     * Gets the value of acm.     *     * @return the value of acm     */    public NacosACMProperties getAcm() {        return acm;    }
    /**     * Sets the acm.     *     * @param acm acm     */    public void setAcm(final NacosACMProperties acm) {        this.acm = acm;    }
    public static class NacosACMProperties {
        private boolean enabled;
        private String endpoint;
        private String namespace;
        private String accessKey;
        private String secretKey;
        /**         * Gets the value of enabled.         *         * @return the value of enabled         */        public boolean isEnabled() {            return enabled;        }
        /**         * Sets the enabled.         *         * @param enabled enabled         */        public void setEnabled(final boolean enabled) {            this.enabled = enabled;        }
        /**         * Gets the value of endpoint.         *         * @return the value of endpoint         */        public String getEndpoint() {            return endpoint;        }
        /**         * Sets the endpoint.         *         * @param endpoint endpoint         */        public void setEndpoint(final String endpoint) {            this.endpoint = endpoint;        }
        /**         * Gets the value of namespace.         *         * @return the value of namespace         */        public String getNamespace() {            return namespace;        }
        /**         * Sets the namespace.         *         * @param namespace namespace         */        public void setNamespace(final String namespace) {            this.namespace = namespace;        }
        /**         * Gets the value of accessKey.         *         * @return the value of accessKey         */        public String getAccessKey() {            return accessKey;        }
        /**         * Sets the accessKey.         *         * @param accessKey accessKey         */        public void setAccessKey(final String accessKey) {            this.accessKey = accessKey;        }
        /**         * Gets the value of secretKey.         *         * @return the value of secretKey         */        public String getSecretKey() {            return secretKey;        }
        /**         * Sets the secretKey.         *         * @param secretKey secretKey         */        public void setSecretKey(final String secretKey) {            this.secretKey = secretKey;        }    }
}

    //DataChangedEventDispatcher.java        @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // 遍历数据变更监听器(一般使用一种数据同步的方式就好了)        for (DataChangedListener listener : listeners) {            // 哪种数据发生变更            switch (event.getGroupKey()) {                                    // 省略了其他逻辑                                    case SELECTOR:   // 选择器信息                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());   // 在我们的案例中，会进入到NacosDataChangedListener进行选择器数据变更处理                    break;         }    }

/** * Use nacos to push data changes. */public class NacosDataChangedListener implements DataChangedListener {    // 选择器信息发生改变    @Override    public void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {        updateSelectorMap(getConfig(NacosPathConstants.SELECTOR_DATA_ID));        switch (eventType) {            case DELETE:                changed.forEach(selector -> {                    List<SelectorData> ls = SELECTOR_MAP                            .getOrDefault(selector.getPluginName(), new ArrayList<>())                            .stream()                            .filter(s -> !s.getId().equals(selector.getId()))                            .sorted(SELECTOR_DATA_COMPARATOR)                            .collect(Collectors.toList());                    SELECTOR_MAP.put(selector.getPluginName(), ls);                });                break;            case REFRESH:            case MYSELF:                SELECTOR_MAP.keySet().removeAll(SELECTOR_MAP.keySet());                changed.forEach(selector -> {                    List<SelectorData> ls = SELECTOR_MAP                            .getOrDefault(selector.getPluginName(), new ArrayList<>())                            .stream()                            .sorted(SELECTOR_DATA_COMPARATOR)                            .collect(Collectors.toList());                    ls.add(selector);                    SELECTOR_MAP.put(selector.getPluginName(), ls);                });                break;            default:                changed.forEach(selector -> {                    List<SelectorData> ls = SELECTOR_MAP                            .getOrDefault(selector.getPluginName(), new ArrayList<>())                            .stream()                            .filter(s -> !s.getId().equals(selector.getId()))                            .sorted(SELECTOR_DATA_COMPARATOR)                            .collect(Collectors.toList());                    ls.add(selector);                    SELECTOR_MAP.put(selector.getPluginName(), ls);                });                break;        }        publishConfig(NacosPathConstants.SELECTOR_DATA_ID, SELECTOR_MAP);    }}

/** * Nacos sync data configuration for spring boot. */@Configuration@ConditionalOnClass(NacosSyncDataService.class)@ConditionalOnProperty(prefix = "shenyu.sync.nacos", name = "url")public class NacosSyncDataConfiguration {
    private static final Logger LOGGER = LoggerFactory.getLogger(NacosSyncDataConfiguration.class);
    /**     * Nacos sync data service.     *     * @param configService     the config service     * @param pluginSubscriber the plugin subscriber     * @param metaSubscribers   the meta subscribers     * @param authSubscribers   the auth subscribers     * @return the sync data service     */    @Bean    public SyncDataService nacosSyncDataService(final ObjectProvider<ConfigService> configService, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,                                           final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {        LOGGER.info("you use nacos sync shenyu data.......");        return new NacosSyncDataService(configService.getIfAvailable(), pluginSubscriber.getIfAvailable(),                metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));    }
    /**     * Nacos config service config service.     *     * @param nacosConfig the nacos config     * @return the config service     * @throws Exception the exception     */    @Bean    public ConfigService nacosConfigService(final NacosConfig nacosConfig) throws Exception {        Properties properties = new Properties();        if (nacosConfig.getAcm() != null && nacosConfig.getAcm().isEnabled()) {            properties.put(PropertyKeyConst.ENDPOINT, nacosConfig.getAcm().getEndpoint());            properties.put(PropertyKeyConst.NAMESPACE, nacosConfig.getAcm().getNamespace());            properties.put(PropertyKeyConst.ACCESS_KEY, nacosConfig.getAcm().getAccessKey());            properties.put(PropertyKeyConst.SECRET_KEY, nacosConfig.getAcm().getSecretKey());        } else {            properties.put(PropertyKeyConst.SERVER_ADDR, nacosConfig.getUrl());            if (StringUtils.isNotBlank(nacosConfig.getNamespace())) {                properties.put(PropertyKeyConst.NAMESPACE, nacosConfig.getNamespace());            }            if (nacosConfig.getUsername() != null) {                properties.put(PropertyKeyConst.USERNAME, nacosConfig.getUsername());            }            if (nacosConfig.getPassword() != null) {                properties.put(PropertyKeyConst.PASSWORD, nacosConfig.getPassword());            }        }        return NacosFactory.createConfigService(properties);    }
    /**     * Http config http config.     *     * @return the http config     */    @Bean    @ConfigurationProperties(prefix = "shenyu.sync.nacos")    public NacosConfig nacosConfig() {        return new NacosConfig();    }}

@Beanpublic SyncDataService nacosSyncDataService(final ObjectProvider<ConfigService> configService, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,                                           final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {        LOGGER.info("you use nacos sync shenyu data.......");        return new NacosSyncDataService(configService.getIfAvailable(), pluginSubscriber.getIfAvailable(),                metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));}

public NacosSyncDataService(final ConfigService configService, final PluginDataSubscriber pluginDataSubscriber,                                final List<MetaDataSubscriber> metaDataSubscribers, final List<AuthDataSubscriber> authDataSubscribers) {
        super(configService, pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);        start();}

    public void start() {        watcherData(NacosPathConstants.PLUGIN_DATA_ID, this::updatePluginMap);        watcherData(NacosPathConstants.SELECTOR_DATA_ID, this::updateSelectorMap);        watcherData(NacosPathConstants.RULE_DATA_ID, this::updateRuleMap);        watcherData(NacosPathConstants.META_DATA_ID, this::updateMetaDataMap);        watcherData(NacosPathConstants.AUTH_DATA_ID, this::updateAuthMap);    }

    protected void watcherData(final String dataId, final OnChange oc) {        Listener listener = new Listener() {            @Override            public void receiveConfigInfo(final String configInfo) {                oc.change(configInfo);            }
            @Override            public Executor getExecutor() {                return null;            }        };        oc.change(getConfigAndSignListener(dataId, listener));        LISTENERS.computeIfAbsent(dataId, key -> new ArrayList<>()).add(listener);    }

    protected void updateSelectorMap(final String configInfo) {        try {            List<SelectorData> selectorDataList = GsonUtils.getInstance().toObjectMapList(configInfo, SelectorData.class).values().stream().flatMap(Collection::stream).collect(Collectors.toList());            selectorDataList.forEach(selectorData -> Optional.ofNullable(pluginDataSubscriber).ifPresent(subscriber -> {                subscriber.unSelectorSubscribe(selectorData);                subscriber.onSelectorSubscribe(selectorData);            }));        } catch (JsonParseException e) {            LOG.error("sync selector data have error:", e);        }    }

/** * 通用插件数据订阅者，负责处理所有插件、选择器和规则信息 * The type Common plugin data subscriber. */public class CommonPluginDataSubscriber implements PluginDataSubscriber {    //......     // 处理选择器数据    @Override    public void onSelectorSubscribe(final SelectorData selectorData) {        subscribeDataHandler(selectorData, DataEventTypeEnum.UPDATE);    }            // 订阅数据处理器，处理数据的更新或删除    private <T> void subscribeDataHandler(final T classData, final DataEventTypeEnum dataType) {        Optional.ofNullable(classData).ifPresent(data -> {            // 插件数据            if (data instanceof PluginData) {                PluginData pluginData = (PluginData) data;                if (dataType == DataEventTypeEnum.UPDATE) { // 更新操作                    // 将数据保存到网关内存                    BaseDataCache.getInstance().cachePluginData(pluginData);                    // 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.handlerPlugin(pluginData));                } else if (dataType == DataEventTypeEnum.DELETE) {  // 删除操作                    // 从网关内存移除数据                    BaseDataCache.getInstance().removePluginData(pluginData);                    // 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.removePlugin(pluginData));                }            } else if (data instanceof SelectorData) {  // 选择器数据                SelectorData selectorData = (SelectorData) data;                if (dataType == DataEventTypeEnum.UPDATE) { // 更新操作                    // 将数据保存到网关内存                    BaseDataCache.getInstance().cacheSelectData(selectorData);                    // 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));                } else if (dataType == DataEventTypeEnum.DELETE) {  // 删除操作                    // 从网关内存移除数据                    BaseDataCache.getInstance().removeSelectData(selectorData);                    // 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.removeSelector(selectorData));                }            } else if (data instanceof RuleData) {  // 规则数据                RuleData ruleData = (RuleData) data;                if (dataType == DataEventTypeEnum.UPDATE) { // 更新操作                    // 将数据保存到网关内存                    BaseDataCache.getInstance().cacheRuleData(ruleData);                    // 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.handlerRule(ruleData));                } else if (dataType == DataEventTypeEnum.DELETE) { // 删除操作                    // 从网关内存移除数据                    BaseDataCache.getInstance().removeRuleData(ruleData);                    // 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.removeRule(ruleData));                }            }        });    }    }

// 将数据保存到网关内存BaseDataCache.getInstance().cacheSelectData(selectorData);// 如果每个插件还有自己的处理逻辑，那么就去处理                    Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));

public final class BaseDataCache {    // 私有变量    private static final BaseDataCache INSTANCE = new BaseDataCache();    // 私有构造器    private BaseDataCache() {    }        /**     * Gets instance.     *  公开方法     * @return the instance     */    public static BaseDataCache getInstance() {        return INSTANCE;    }        /**    *  缓存选择器数据的Map     * pluginName -> SelectorData.     */    private static final ConcurrentMap<String, List<SelectorData>> SELECTOR_MAP = Maps.newConcurrentMap();        public void cacheSelectData(final SelectorData selectorData) {        Optional.ofNullable(selectorData).ifPresent(this::selectorAccept);    }           /**     * cache selector data.     * 缓存选择器数据     * @param data the selector data     */    private void selectorAccept(final SelectorData data) {        String key = data.getPluginName();        if (SELECTOR_MAP.containsKey(key)) { // 更新操作，先删除再插入            List<SelectorData> existList = SELECTOR_MAP.get(key);            final List<SelectorData> resultList = existList.stream().filter(r -> !r.getId().equals(data.getId())).collect(Collectors.toList());            resultList.add(data);            final List<SelectorData> collect = resultList.stream().sorted(Comparator.comparing(SelectorData::getSort)).collect(Collectors.toList());            SELECTOR_MAP.put(key, collect);        } else {  // 新增操作，直接放到Map中            SELECTOR_MAP.put(key, Lists.newArrayList(data));        }    }    }

/** * The type Nacos data init. */public class NacosDataInit implements CommandLineRunner {
    private static final Logger LOG = LoggerFactory.getLogger(NacosDataInit.class);
    private final ConfigService configService;
    private final SyncDataService syncDataService;
    /**     * Instantiates a new Nacos data init.     * @param configService the nacos config service     * @param syncDataService the sync data service     */    public NacosDataInit(final ConfigService configService, final SyncDataService syncDataService) {        this.configService = configService;        this.syncDataService = syncDataService;    }
    @Override    public void run(final String... args) {        String pluginDataId = NacosPathConstants.PLUGIN_DATA_ID;        String authDataId = NacosPathConstants.AUTH_DATA_ID;        String metaDataId = NacosPathConstants.META_DATA_ID;        if (dataIdNotExist(pluginDataId) && dataIdNotExist(authDataId) && dataIdNotExist(metaDataId)) {            syncDataService.syncAll(DataEventTypeEnum.REFRESH);        }    }
    private boolean dataIdNotExist(final String pluginDataId) {        try {            String group = NacosPathConstants.GROUP;            long timeout = NacosPathConstants.DEFAULT_TIME_OUT;            return configService.getConfig(pluginDataId, group, timeout) == null;        } catch (NacosException e) {            LOG.error("Get data from nacos error.", e);            throw new ShenyuException(e.getMessage());        }    }}

@Servicepublic class SyncDataServiceImpl implements SyncDataService {    // 事件发布    private final ApplicationEventPublisher eventPublisher;         /***     * 全量数据同步     * @param type the type     * @return     */    @Override    public boolean syncAll(final DataEventTypeEnum type) {        // 同步认证信息        appAuthService.syncData();        // 同步插件信息        List<PluginData> pluginDataList = pluginService.listAll();        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, type, pluginDataList));        // 同步选择器信息        List<SelectorData> selectorDataList = selectorService.listAll();        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, type, selectorDataList));        // 同步规则信息        List<RuleData> ruleDataList = ruleService.listAll();        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.RULE, type, ruleDataList));        // 同步元数据信息        metaDataService.syncData();        return true;    }    }

@Validated@RequiredArgsConstructor@RestController@RequestMapping("/selector")public class SelectorController {        @PostMapping("")    public ShenyuAdminResult createSelector(@Valid @RequestBody final SelectorDTO selectorDTO) { // @Valid 数校验        // 添加或更新数据        Integer createCount = selectorService.createOrUpdate(selectorDTO);        // 返回结果信息        return ShenyuAdminResult.success(ShenyuResultMessage.CREATE_SUCCESS, createCount);    }        // ......}

@RequiredArgsConstructor@Servicepublic class SelectorServiceImpl implements SelectorService {    // 负责事件发布的eventPublisher    private final ApplicationEventPublisher eventPublisher;        @Override    @Transactional(rollbackFor = Exception.class)    public int createOrUpdate(final SelectorDTO selectorDTO) {        int selectorCount;        // 构建数据 DTO --> DO        SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);        List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();        // 判断是添加还是更新        if (StringUtils.isEmpty(selectorDTO.getId())) {            // 插入选择器数据            selectorCount = selectorMapper.insertSelective(selectorDO);            // 插入选择器中的条件数据            selectorConditionDTOs.forEach(selectorConditionDTO -> {                selectorConditionDTO.setSelectorId(selectorDO.getId());                selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));            });            // check selector add            // 权限检查            if (dataPermissionMapper.listByUserId(JwtUtils.getUserInfo().getUserId()).size() > 0) {                DataPermissionDTO dataPermissionDTO = new DataPermissionDTO();                dataPermissionDTO.setUserId(JwtUtils.getUserInfo().getUserId());                dataPermissionDTO.setDataId(selectorDO.getId());                dataPermissionDTO.setDataType(AdminConstants.SELECTOR_DATA_TYPE);                dataPermissionMapper.insertSelective(DataPermissionDO.buildPermissionDO(dataPermissionDTO));            }
        } else {            // 更新数据，先删除再新增            selectorCount = selectorMapper.updateSelective(selectorDO);            //delete rule condition then add            selectorConditionMapper.deleteByQuery(new SelectorConditionQuery(selectorDO.getId()));            selectorConditionDTOs.forEach(selectorConditionDTO -> {                selectorConditionDTO.setSelectorId(selectorDO.getId());                SelectorConditionDO selectorConditionDO = SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO);                selectorConditionMapper.insertSelective(selectorConditionDO);            });        }        // 发布事件        publishEvent(selectorDO, selectorConditionDTOs);
        // 更新upstream        updateDivideUpstream(selectorDO);        return selectorCount;    }            // ......    }

       private void publishEvent(final SelectorDO selectorDO, final List<SelectorConditionDTO> selectorConditionDTOs) {        // 找到选择器对应的插件        PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());        // 构建条件数据        List<ConditionData> conditionDataList =                selectorConditionDTOs.stream().map(ConditionTransfer.INSTANCE::mapToSelectorDTO).collect(Collectors.toList());        // 发布变更数据        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,                Collections.singletonList(SelectorDO.transFrom(selectorDO, pluginDO.getName(), conditionDataList))));    }

public class DataChangedEvent extends ApplicationEvent {//......}

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
    //......    }

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
  /**     * 有数据变更时，调用此方法     * @param event     */    @Override    @SuppressWarnings("unchecked")    public void onApplicationEvent(final DataChangedEvent event) {        // 遍历数据变更监听器(一般使用一种数据同步的方式就好了)        for (DataChangedListener listener : listeners) {            // 哪种数据发生变更            switch (event.getGroupKey()) {                case APP_AUTH: // 认证信息                    listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());                    break;                case PLUGIN:  // 插件信息                    listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());                    break;                case RULE:    // 规则信息                    listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());                    break;                case SELECTOR:   // 选择器信息                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());                    break;                case META_DATA:  // 元数据                    listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());                    break;                default:  // 其他类型，抛出异常                    throw new IllegalStateException("Unexpected value: " + event.getGroupKey());            }        }    }    }