One post tagged with "nacos"

Nacos数据同步源码分析

2024年12月5日 · One min read

Apache ShenYu Contributor

Apache ShenYu 是一个异步的，高性能的，跨语言的，响应式的 API 网关。

在ShenYu网关中，数据同步是指，当在后台管理系统中，数据发送了更新后，如何将更新的数据同步到网关中。Apache ShenYu 网关当前支持ZooKeeper、WebSocket、Http长轮询、Nacos 、Etcd 和 Consul 进行数据同步。本文的主要内容是基于Nacos的数据同步源码分析。

本文基于shenyu-2.4.0版本进行源码分析，官网的介绍请参考数据同步原理。

1. 关于Nacos#

Nacos 平台用于动态服务发现，以及配置和服务管理。 Shenyu网关可选择使用Nacos进行数据同步。

2. Admin数据同步#

我们从一个实际案例进行源码追踪，比如在后台管理系统中，对Divide插件中的一条选择器数据进行更新，将权重更新为90：

2.1 接收数据#

SelectorController.updateSelector()

进入SelectorController类中的updateSelector()方法，它负责数据的校验，添加或更新数据，返回结果信息。

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

2.2 处理数据#

SelectorServiceImpl.createOrUpdate()

在SelectorServiceImpl类中通过createOrUpdate()方法完成数据的转换，保存到数据库，发布事件，更新upstream。

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

在Service类完成数据的持久化操作，即保存数据到数据库，这个比较简单，就不深入追踪了。关于更新upstream操作，放到后面对应的章节中进行分析，重点关注发布事件的操作，它会执行数据同步。

publishEvent()方法的逻辑是：找到选择器对应的插件，构建条件数据，发布变更数据。

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

发布变更数据通过eventPublisher.publishEvent()完成，这个eventPublisher对象是一个ApplicationEventPublisher类，这个类的全限定名是org.springframework.context.ApplicationEventPublisher。看到这儿，我们知道了发布数据是通过Spring相关的功能来完成的。

关于ApplicationEventPublisher：
当有状态发生变化时，发布者调用 ApplicationEventPublisher 的 publishEvent 方法发布一个事件，Spring容器广播事件给所有观察者，调用观察者的 onApplicationEvent 方法把事件对象传递给观察者。调用 publishEvent方法有两种途径，一种是实现接口由容器注入 ApplicationEventPublisher 对象然后调用其方法，另一种是直接调用容器的方法，两种方法发布事件没有太大区别。
ApplicationEventPublisher：发布事件；
ApplicationEvent：Spring 事件，记录事件源、时间和数据；
ApplicationListener：事件监听者，观察者；

在Spring的事件发布机制中，有三个对象，

一个是发布事件的ApplicationEventPublisher，在ShenYu中通过构造器注入了一个eventPublisher。

另一个对象是ApplicationEvent，在ShenYu中通过DataChangedEvent继承了它，表示事件对象。

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

最后一个是 ApplicationListener，在ShenYu中通过DataChangedEventDispatcher类实现了该接口，作为事件的监听者，负责处理事件对象。

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

2.3 分发数据#

DataChangedEventDispatcher.onApplicationEvent()

当事件发布完成后，会自动进入到DataChangedEventDispatcher类中的onApplicationEvent()方法，进行事件处理。

@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());            }        }    }    }

当有数据变更时，调用onApplicationEvent方法，然后遍历所有数据变更监听器，判断是哪种数据类型，交给相应的数据监听器进行处理。

ShenYu将所有数据进行了分组，一共是五种：认证信息、插件信息、规则信息、选择器信息和元数据。

这里的数据变更监听器（DataChangedListener），就是数据同步策略的抽象，它的具体实现有：

这几个实现类就是当前ShenYu支持的同步策略：

WebsocketDataChangedListener：基于websocket的数据同步；
ZookeeperDataChangedListener：基于zookeeper的数据同步；
ConsulDataChangedListener：基于consul的数据同步；
EtcdDataDataChangedListener：基于etcd的数据同步；
HttpLongPollingDataChangedListener：基于http长轮询的数据同步；
NacosDataChangedListener：基于nacos的数据同步；

既然有这么多种实现策略，那么如何确定使用哪一种呢？

因为本文是基于Nacos的数据同步源码分析，所以这里以NacosDataChangedListener为例，分析它是如何被加载并实现的。

通过查看对NacosDataChangedListener类的调用，可以发现，它是在DataSyncConfiguration类进行配置的。

/** * 数据同步配置类 * 通过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);        }    }         //省略了其他代码......}

这个配置类是通过SpringBoot条件装配类实现的。在NacosListener类上面有几个注解：

@Configuration：配置文件，应用上下文；
@ConditionalOnProperty(prefix = "shenyu.sync.nacos", name = "url")：属性条件判断，满足条件，该配置类才会生效。也就是说，当我们有如下配置时，就会采用nacos进行数据同步。
```
shenyu:    sync:     nacos:          url: localhost:8848
```
@Import(NacosConfiguration.class)：导入另一个配置类NacosConfiguration，NacosConfiguration提供了一个方法ConfigService nacosConfigService(final NacosProperties nacosProp)，将Nacos属性转换为ConfigService类型的bean，而Nacos属性是通过@EnableConfigurationProperties(NacosProperties.class) 导入的。我们先看ConfigService类型的bean定义。再分析属性配置类和对应的属性配置文件。

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

这个方法主要分成两步，第一步根据是否使用了aliyun的ACM服务，从NacosProperties中获取不同的nacos路径和鉴权信息，第二步根据获取到的这些属性，使用Nacos官方的工厂方法，使用反射的方式，创建configService。

接下来，让我们分析一下Nacos的属性配置和对应的配置文件。

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

当我们在配置文件中配置了shenyu.sync.nacos.url属性时，将采用nacos进行数据同步，此时配置类NacosListener会生效，并生成NacosDataChangedListener和NacosDataInit类型的bean。

生成NacosDataChangedListener类型的bean，nacosDataChangedListener，这个bean将ConfigService类型的bean作为成员变量，ConfigService是nacos官方提供的api，当nacosDataChangedListener监听到事件时，进行回调操作，可以通过该api直接与nacos服务器交互，修改配置。
生成NacosDataInit类型的bean，nacosDataInit，这个bean将beanconfigService和beansyncDataService作为成员变量，调用Nacos的api configService判断配置是否未初始化，未初始化则调用syncDataService进行刷新操作，将在下文详述。根据上文所述，在事件处理方法onApplicationEvent()中，会触发相应的listener的操作。在我们的案例中，是对一条选择器数据进行更新，数据同步采用的是nacos，所以，代码会进入到NacosDataChangedListener进行选择器数据变更处理。

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

2.4 Nacos数据变更监听器#

NacosDataChangedListener.onSelectorChanged()
在onSelectorChanged()方法中，判断操作类型，是刷新同步还是更新或创建同步。根据当前选择器数据信息判断节点是否在nacos中。

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

这部分是核心。changed表示需更新的SelectorData列表，eventType表示事件类型。SELECTOR_MAP的类型是ConcurrentMap<String, List<SelectorData>>，该map的key为selector所属的plugin的名称，value为该plugin下的selector列表。NacosPathConstants.SELECTOR_DATA_ID的值为shenyu.selector.json。操作步骤如下，第一步，使用getConfig方法调用Nacos的api，从Nacos获取group为shenyu.selector.json的配置信息，updateSelectorMap方法使用这些配置信息更新SELECTOR_MAP，这样就同步到了Nacos上最新的selector信息。第二步，再根据事件类型来更新SELECTOR_MAP，最后使用publishConfig方法，调用Nacos的api，将Nacos上，group为shenyu.selector.json的配置进行全量替换。

只要将变动的数据正确写入到Nacos上，admin这边的操作就执行完成了。

在我们当前的案例中，对Divide插件中的一条选择器数据进行更新，将权重更新为90，就会对图中的特定节点更新。

我们用时序图将上面的更新流程串联起来。

3. 网关数据同步#

假设ShenYu网关已经在正常运行，使用的数据同步方式也是nacos。那么当在admin端更新选择器数据后，并且向nacos发送了变更的数据，那网关是如何接收并处理数据的呢？接下来我们就继续进行源码分析，一探究竟。

3.1 `NacosSyncDataService`接收数据#

网关是通过NacosSyncDataService对nacos进行监听并获取数据更新的，但是在这部分内容之前，我们先看一下NacosSyncDataService类型的bean是如何生成的。答案是在Spring配置类NacosSyncDataConfiguration中定义的。我们看到NacosSyncDataConfiguration类上的注解，@ConditionalOnProperty(prefix = "shenyu.sync.nacos", name = "url")，这个注解我们在上文对ShenYu的Admin端中的NacosListener类进行分析时看到过，是一个属性条件判断，满足条件，该配置类才会生效。也就是说，当我们在Shenyu网关端有如下配置时，就表示Shenyu网关端采用nacos进行数据同步，NacosSyncDataConfiguration这个配置类生效。

shenyu:    sync:     nacos:          url: localhost:8848

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

我们重点关注一下上面代码中nacosSyncDataService这个bean的生成：

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

是直接调用NacosSyncDataService的构造方法new了一个该类型的对象。我们继续看构造方法：

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

可以看到，在构造方法中调用了start方法，并且通过watcherData方法创建了监听器，并且关联了回调函数oc，由于我们正在分析selector类型组件的变化，对应的回调函数是updateSelectorMap。这个回调函数用于处理数据。

3.2 处理数据#

NacosCacheHandler.updateSelectorMap()

经过判空逻辑之后，缓存选择器数据的操作又交给了PluginDataSubscriber处理。

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

PluginDataSubscriber是一个接口，它只有一个CommonPluginDataSubscriber实现类，负责处理插件、选择器和规则数据。

3.3 通用插件数据订阅者#

PluginDataSubscriber.onSelectorSubscribe()

它没有其他逻辑，直接调用subscribeDataHandler()方法。在方法中，更具数据类型（插件、选择器或规则），操作类型（更新或删除），去执行不同逻辑。

/** * 通用插件数据订阅者，负责处理所有插件、选择器和规则信息 * 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));                }            }        });    }    }

3.4 数据缓存到内存#

那么更新一条选择器数据，会进入下面的逻辑：

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

一是将数据保存到网关的内存中。BaseDataCache是最终缓存数据的类，通过单例模式实现。选择器数据就存到了SELECTOR_MAP这个Map中。在后续使用的时候，也是从这里拿数据。

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

二是如果每个插件还有自己的处理逻辑，那么就去处理。通过idea编辑器可以看到，当新增一条选择器后，有如下的插件还有处理。这里我们就不再展开了。

经过以上的源码追踪，并通过一个实际的案例，在admin端新增更新一条选择器数据，就将nacos数据同步的流程分析清楚了。

我们还是通过时序图将网关端的数据同步流程串联一下：

数据同步的流程已经分析完了，为了不让同步流程被打断，在分析过程中就忽略了其他逻辑。网关同步操作初始化的流程在NacosSyncDataService的start方法中，我们在上文分析网关数据同步时分析过了，下面分析Admin的同步数据初始化。

4. Admin同步数据初始化#

admin端，NacosDataInit类型的bean，在NacosListener中进行定义和生成，如果admin的配置中指定了使用nacos进行数据同步，当admin启动后，会将当前的数据信息全量同步到nacos中，实现逻辑如下：


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

判断nacos中是否存在数据，如果不存在，则进行同步。

NacosDataInit实现了CommandLineRunner接口。它是springboot提供的接口，会在所有 Spring Beans初始化之后执行run()方法，常用于项目中初始化的操作。

SyncDataService.syncAll()

从数据库查询数据，然后进行全量数据同步，所有的认证信息、插件信息、选择器信息、规则信息和元数据信息。主要是通过eventPublisher发布同步事件。这里就跟前面提到的同步逻辑就又联系起来了，eventPublisher通过publishEvent()发布完事件后，有ApplicationListener执行事件变更操作，在ShenYu中就是前面提到的DataChangedEventDispatcher。

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

5. 总结#

本文通过一个实际案例，对nacos的数据同步原理进行了源码分析。涉及到的主要知识点如下：

基于nacos的数据同步，主要是通过watch机制实现；
通过Spring完成事件发布和监听；
通过抽象DataChangedListener接口，支持多种同步策略，面向接口编程；
使用单例设计模式实现缓存数据类BaseDataCache；
通过SpringBoot的条件装配和starter加载机制实现配置类的加载。