1.準備

按照昨天的帖子（【Soul原始碼閱讀】14.soul-admin 與 soul-bootstrap 同步機制之 nacos 決議（上）），把專案啟動起來，soul-admin、soul-bootstrap 和 soul-examples-http 三個專案，當然還有依賴的 MySQL 和 nacos，

2.頁面操作

我們以修改 selector 為例，

修改前，在 nacos 里看下 Data Id 為 "soul.selector.json" 的資料如下：

{"divide":[{"id":"1354790660314062848","pluginId":"5","pluginName":"divide","name":"/http","matchMode":0,"type":1,"sort":1,"enabled":true,"loged":true,"continued":true,"handle":"[{\"upstreamHost\":\"localhost\",\"protocol\":\"http://\",\"upstreamUrl\":\"10.0.0.5:8188\",\"weight\":50,\"status\":true,\"timestamp\":0,\"warmup\":0}]","conditionList":[{"paramType":"uri","operator":"match","paramName":"/","paramValue":"/http/**"}]}]}

格式化如下：（可以看到 divide 插件里有一個 http 的 selector）

{
"divide": [
{
"id": "1354790660314062848",
"pluginId": "5",
"pluginName": "divide",
"name": "/http",
"matchMode": 0,
"type": 1,
"sort": 1,
"enabled": true,
"loged": true,
"continued": true,
"handle": "[{\"upstreamHost\":\"localhost\",\"protocol\":\"http://\",\"upstreamUrl\":\"10.0.0.5:8188\",\"weight\":50,\"status\":true,\"timestamp\":0,\"warmup\":0}]",
"conditionList": [
{
"paramType": "uri",
"operator": "match",
"paramName": "/",
"paramValue": "/http/**"
}
]
}
]
}

在 handle 欄位中，有具體的業務系統資訊，我們把權重 weight 從50改到80，

再次查看 nacos 中的配置資料：

{"divide":[{"id":"1354790660314062848","pluginId":"5","pluginName":"divide","name":"/http","matchMode":0,"type":1,"sort":1,"enabled":true,"loged":true,"continued":true,"handle":"[{\"upstreamHost\":\"localhost\",\"protocol\":\"http://\",\"upstreamUrl\":\"10.0.0.5:8188\",\"weight\":\"80\",\"status\":true,\"timestamp\":\"0\",\"warmup\":\"0\"}]","conditionList":[{"paramType":"uri","operator":"match","paramName":"/","paramValue":"/http/**"}]}]}

很顯然，weight 改成了 80，

下面我們看下是如何實作的，

3.跟蹤原始碼

3.1 soul-admin 端

F12 可以看到修改 selector 時發送了一個 PUT 請求 http://localhost:9095/selector/1354790660314062848

很容易就找到了 SelectorController：

// SelectorController.java
    /**
     * update Selector.
     *
     * @param id          primary key.
     * @param selectorDTO selector.
     * @return {@linkplain SoulAdminResult}
     */
    @PutMapping("/{id}")
    public SoulAdminResult updateSelector(@PathVariable("id") final String id, @RequestBody final SelectorDTO selectorDTO) {
        Objects.requireNonNull(selectorDTO);
        selectorDTO.setId(id);
        Integer updateCount = selectorService.createOrUpdate(selectorDTO);
        return SoulAdminResult.success(SoulResultMessage.UPDATE_SUCCESS, updateCount);
    }

邏輯都在 createOrUpdate 方法里：

// SelectorServiceImpl.java
    /**
     * create or update selector.
     *
     * @param selectorDTO {@linkplain SelectorDTO}
     * @return rows
     */
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public int createOrUpdate(final SelectorDTO selectorDTO) {
        int selectorCount;
        SelectorDO selectorDO = SelectorDO.buildSelectorDO(selectorDTO);
        List<SelectorConditionDTO> selectorConditionDTOs = selectorDTO.getSelectorConditions();
        if (StringUtils.isEmpty(selectorDTO.getId())) {
            // 資料沒有 ID，走新增邏輯
            selectorCount = selectorMapper.insertSelective(selectorDO);
            selectorConditionDTOs.forEach(selectorConditionDTO -> {
                selectorConditionDTO.setSelectorId(selectorDO.getId());
                selectorConditionMapper.insertSelective(SelectorConditionDO.buildSelectorConditionDO(selectorConditionDTO));
            });
        } else {
            // 資料有 ID，走更新邏輯
            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);
        updateDivideUpstream(selectorDO);
        return selectorCount;
    }

前半部分都是資料庫操作，就不過多解釋了，

publishEvent 方法，發布事件：

// SelectorServiceImpl.java
    private void publishEvent(final SelectorDO selectorDO, final List<SelectorConditionDTO> selectorConditionDTOs) {
        // 從資料庫獲取 selector 對應的 pulugin 資訊
        PluginDO pluginDO = pluginMapper.selectById(selectorDO.getPluginId());
        // 轉換資料格式
        List<ConditionData> conditionDataList =
                selectorConditionDTOs.stream().map(ConditionTransfer.INSTANCE::mapToSelectorDTO).collect(Collectors.toList());
        // publish change event.
        // 發布事件
        eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,
                Collections.singletonList(SelectorDO.transFrom(selectorDO, pluginDO.getName(), conditionDataList))));
    }

通過發布事件、監聽事件，解耦代碼呼叫，

監聽事件代碼：

// DataChangedEventDispatcher.java
@Override
    @SuppressWarnings("unchecked")
    public void onApplicationEvent(final DataChangedEvent event) {
        for (DataChangedListener listener : listeners) {
            // 根據 groupKey 型別匹配不同的邏輯，我們這里是 SELECTOR
            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());
            }
        }
    }

debug 跟蹤到這里時，有2個監聽器，我們使用的 nacos 同步策略，關注 NacosDataChangedListener 即可，

// NacosDataChangedListener.java
@Override
    public void onSelectorChanged(final List<SelectorData> changed, final DataEventTypeEnum eventType) {
        // 從 nacos 獲取配置資訊，并更新本地快取 SELECTOR_MAP 中資料
        updateSelectorMap(getConfig(NacosPathConstants.SELECTOR_DATA_ID));
        // 根據不同事件型別，執行不同邏輯，我們這里型別是 UPDATE，執行 default 邏輯
        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:
                // 這里又對 SELECTOR_MAP 更新了一遍，沒太搞明白，有誰搞清楚了可以跟我說下嗎？
                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;
        }
        // 發布配置，將配置資訊保存到 nacos
        publishConfig(NacosPathConstants.SELECTOR_DATA_ID, SELECTOR_MAP);
    }


    @SneakyThrows
    private String getConfig(final String dataId) {
        // 從 nacos 獲取配置資訊
        String config = configService.getConfig(dataId, NacosPathConstants.GROUP, NacosPathConstants.DEFAULT_TIME_OUT);
        return StringUtils.hasLength(config) ? config : NacosPathConstants.EMPTY_CONFIG_DEFAULT_VALUE;
    }

    // 更新本地快取資料
    private void updateSelectorMap(final String configInfo) {
        JsonObject jo = GsonUtils.getInstance().fromJson(configInfo, JsonObject.class);
        Set<String> set = new HashSet<>(SELECTOR_MAP.keySet());
        for (Entry<String, JsonElement> e : jo.entrySet()) {
            set.remove(e.getKey());
            List<SelectorData> ls = new ArrayList<>();
            e.getValue().getAsJsonArray().forEach(je -> ls.add(GsonUtils.getInstance().fromJson(je, SelectorData.class)));
            SELECTOR_MAP.put(e.getKey(), ls);
        }
        SELECTOR_MAP.keySet().removeAll(set);
    }

    @SneakyThrows
    private void publishConfig(final String dataId, final Object data) {
        configService.publishConfig(dataId, NacosPathConstants.GROUP, GsonUtils.getInstance().toJson(data));
    }

3.2 soul-bootstrap 端

這里一旦把資料保存到 nacos，就會觸發昨天分析的監聽器，呼叫 receiveConfigInfo 方法，

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

呼叫 oc.change 方法，就回到了上一篇文章中分析的5個方法 updateXxxMap，更新記憶體資料，

好了，到這里就把 nacos 同步策略大體流程分析完了，同步策略涉及的內容有點兒多，有點兒雜，明天整理一篇整合文章，

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/254911.html

標籤：其他

上一篇：快速排序(雙指標法、挖坑法、前后指標法)遞回、非遞回實作

下一篇：Redis常見問題

【Soul原始碼閱讀】15.soul-admin 與 soul-bootstrap 同步機制之 nacos 決議（下）

1.準備

2.頁面操作

3.跟蹤原始碼

3.1 soul-admin 端

3.2 soul-bootstrap 端