天天看點
傳回

Soul 網關 Nacos 資料同步源碼解析

學習目标：

學習Soul 網關 Nacos 資料同步源碼解析

學習内容：

  1. 環境配置
  2. Soul 網關 Nacos 資料同步基本概念
  3. 源碼分析

學習時間：2020年1月28号 早7點

學習産出：

環境配置：

  1. 引入依賴

    在 soul-bootstrap 項目的 pom.xml 檔案中引入了 soul-spring-boot-starter-sync-data-nacos 這個 starter 。

<dependency>
          <groupId>org.dromara</groupId>
           <artifactId>soul-spring-boot-starter-sync-data-nacos</artifactId>
           <version>${last.version}</version>
     </dependency>
  1. 環境配置

在 soul-bootstrap 項目的 application-local.yml 檔案中，配置

soul :
	 sync:
	    nacos:
	         url: localhost:8848
	         namespace: 1c10d748-af86-43b9-8265-75f487d20c6c
	         acm:
	           enabled: false
	           endpoint: acm.aliyun.com
	           namespace:
	           accessKey:
	           secretKey:
	           #url: 配置成你的nacos位址，叢集環境請使用（,）分隔。

在 soul-admin 項目中的 application.yml 檔案中配置 nacos 同步的相關配置如下：

soul:
  sync:
    nacos:
      url: localhost:8848
      namespace: 7aba82c9-426b-4d72-89c9-060d9af63e14
       # 此處為前文中提到的命名空間ID或命名空間名稱 soul

Nacos 的關鍵特性

服務發現和服務健康監測

動态配置服務

動态 DNS 服務

服務及其中繼資料管理

Nacos 中的幾個概念

命名空間(Namespace)： 不同環境的配置隔離

配置分組(Group)： 不同的服務可以歸類到同一分組。一般将一個項目的配置分到一組

配置集(Data ID) ： 一個配置檔案通常就是一個配置集

  1. nacos同步政策-源碼追蹤

    從 soul-bootstrap 開始追蹤

    NacosSyncDataConfiguration 這個類加載了 soul.sync.nacos 這段配置。該類建立了 nacosSyncDataService 。

@Bean
public SyncDataService nacosSyncDataService(final ObjectProvider<ConfigService> configService, final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
                                            final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
    log.info("you use nacos sync soul 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(PLUGIN_DATA_ID, this::updatePluginMap);
    watcherData(SELECTOR_DATA_ID, this::updateSelectorMap);
    watcherData(RULE_DATA_ID, this::updateRuleMap);
    watcherData(META_DATA_ID, this::updateMetaDataMap);
    watcherData(AUTH_DATA_ID, this::updateAuthMap);
}

進入 watcherData 方法，此時我們會進入一個重要的類 NacosCacheHandler 。

// 即便對 lambda 不熟悉，我們也可以大緻推測，這裡的功能應該是通過調用updateXxxMap方法，對中繼資料、插件、選擇器等進行監聽
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 interface OnChange {
    void change(String changeData);
}

updatePluginMap 方法:

protected void updatePluginMap(final String configInfo) {
    try {
        // Fix bug #656(https://github.com/dromara/soul/issues/656) 此處解決了一個 nacos data sync error 的 bug
        List<PluginData> pluginDataList = new ArrayList<>(GsonUtils.getInstance().toObjectMap(configInfo, PluginData.class).values());
        pluginDataList.forEach(pluginData -> Optional.ofNullable(pluginDataSubscriber).ifPresent(subscriber -> {
            subscriber.unSubscribe(pluginData);
            subscriber.onSubscribe(pluginData);
        }));
    } catch (JsonParseException e) {
        log.error("sync plugin data have error:", e);
    }
}

追蹤一下 unSubscribe(pluginData) 和 onSubscribe(pluginData) ，進入 CommonPluginDataSubscriber 中：

@Override
public void onSubscribe(final PluginData pluginData) {
    subscribeDataHandler(pluginData, DataEventTypeEnum.UPDATE);
}

@Override
public void unSubscribe(final PluginData pluginData) {
    subscribeDataHandler(pluginData, DataEventTypeEnum.DELETE);
}

// 根據傳入的資料類型ID，和資料處理類型，調用相應的處理函數，看具體是執行update更新還是delete删除
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));
            }
        }
    });
}

admin追蹤：

是 DataSyncConfiguration 這個類加載了 soul.sync.nacos 這段配置。該類建立了 NacosDataChangedListener 。

/**
 * The type Nacos listener.
 */
@Configuration
@ConditionalOnProperty(prefix = "soul.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);
    }
}
           
@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);
    }
}
           
// type 是 REFRESH
@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;
}
           
public static void send(final String message, final DataEventTypeEnum type) {
    if (StringUtils.isNotBlank(message)) {
        if (DataEventTypeEnum.MYSELF == type) {
            Session session = (Session) ThreadLocalUtil.get(SESSION_KEY);
            if (session != null) {
                sendMessageBySession(session, message);
            }
        } else {
            SESSION_SET.forEach(session -> sendMessageBySession(session, message));
        }
    }
}
上一篇: 擷取網關Ip C#
下一篇: 三. SpringCloud服務注冊與發現