Apache 神禹（shenyu）源码阅读（二）——Admin 向 Gateway 的数据同步（Gateway 端）

2024-02-16 09:52:01
开发
29

源码版本：2.6.1

前言

上一篇Apache 神禹（shenyu）源码阅读（一）——Admin向Gateway的数据同步（Admin端）写了Admin 端在接收到程序员对 Divide 插件的选择器 Selector 作出新增操作时，Admin 端是如何将要同步的数据发布给 Gateway 端的。

本篇介绍 Gateway 端是如何接收 Admin 端发布的数据的。

本文介绍的数据同步（sync data）在 Shenyu 架构图中的位置

在这里插入图片描述

正文

1. Gateway 端通过网络接收 Admin 端要同步的数据

ShenyuWebsocketClient.onMessage()
由 Admin 端的 WebsocketCollector.send() 通过网络发送数据后（上一篇的内容），Gateway 端的 ShenyuWebsocketClient.onMessage() 收到数据，onMessage() 是 Spring 框架抽象类 WebSocketClient 的一个方法，在 ShenyuWebsocketClient 中实现了这个方法。

public final class ShenyuWebsocketClient extends WebSocketClient {
   
	// ...
    @Override
    public void onMessage(final String result) {
   
        handleResult(result);
    }
    private void handleResult(final String result) {
   
        // 1. 打印日志
        LOG.info("handleResult({})", result);
        // 2. 调用 Gson 包，将 Json 字符串转换为 WebsocketData
        WebsocketData<?> websocketData = GsonUtils.getInstance().fromJson(result, WebsocketData.class);
        // 3. 因为我们是新增的 Selector，所以这里 groupEnum 为 ConfigGroupEnum.SELECTOR
        ConfigGroupEnum groupEnum = ConfigGroupEnum.acquireByName(websocketData.getGroupType());
        // 4. 事件是 UPDATE
        String eventType = websocketData.getEventType();
        // 5. 再转成 Json 字符串
        String json = GsonUtils.getInstance().toJson(websocketData.getData());
        // 6. 交给 WebsocketDataHandler 处理数据
        websocketDataHandler.executor(groupEnum, json, eventType);
    }
}

ShenyuWebsocketClient.handleResult()

如上面那段代码，
1. 打印日志
2. 调用 Gson 包，将 Json 字符串转换为 WebsocketData
3. 因为我们是新增的 Selector，所以这里 groupEnum 为 ConfigGroupEnum.SELECTOR
4. 事件是 UPDATE
5. 再转成 Json 字符串
6. 交给 WebsocketDataHandler 处理数据
WebsocketDataHandler.executor()

WebsocketDataHandler 的一个 EnumMap 类型的成员变量存储了 ConfigGroupEnum -> DataHandler 的映射。在 executor 方法里拿到 ConfigGroupEnum 对应的 DataHandler 去处理数据

public class WebsocketDataHandler {
   
	// ...
    private static final EnumMap<ConfigGroupEnum, DataHandler> ENUM_MAP = new EnumMap<>(ConfigGroupEnum.class);    

	public void executor(final ConfigGroupEnum type, final String json, final String eventType) {
   
        ENUM_MAP.get(type).handle(json, eventType);
    }
}

2 交由 SelectorDataHandler 处理数据

DataHandler.handle

DataHandler 是个接口：

public interface DataHandler {
   

    /**
     * Handle.
     *
     * @param json  the data for json
     * @param eventType the event type
     */
    void handle(String json, String eventType);
}

其继承关系如下图：
在这里插入图片描述

AbstractDataHandler.handle()

这里 handle() 用到了一个设计模式——模板方法，里面用到的方法都是交由子类根据自己的逻辑去实现
事件类型为 UPDATE 和 CREATE 的事件都由 doUpdate 方法处理

public abstract class AbstractDataHandler<T> implements DataHandler {
   

    /**
     * Convert list.
     *
     * @param json the json
     * @return the list
     */
    protected abstract List<T> convert(String json);

    /**
     * Do refresh.
     *
     * @param dataList the data list
     */
    protected abstract void doRefresh(List<T> dataList);

    /**
     * Do update.
     *
     * @param dataList the data list
     */
    protected abstract void doUpdate(List<T> dataList);

    /**
     * Do delete.
     *
     * @param dataList the data list
     */
    protected abstract void doDelete(List<T> dataList);

    @Override
    public void handle(final String json, final String eventType) {
   
        List<T> dataList = convert(json);

        if (CollectionUtils.isEmpty(dataList)) {
   
            return;
        }

        DataEventTypeEnum eventTypeEnum = DataEventTypeEnum.acquireByName(eventType);
        switch (eventTypeEnum) {
   
            case REFRESH:
            case MYSELF:
                doRefresh(dataList);
                break;
            case UPDATE:
            case CREATE:
            	// 事件类型为 UPDATE 和 CREATE 的事件都由 doUpdate 方法处理
                doUpdate(dataList);
                break;
            case DELETE:
                doDelete(dataList);
                break;
            default:
                break;
        }
    }
}

SelectorDataHandler.doUpdate()

由插件数据订阅者 pluginDataSubscriber 去完成 Selector 数据的订阅和处理

 public class SelectorDataHandler extends AbstractDataHandler<SelectorData> {
   
    // ...
    private final PluginDataSubscriber pluginDataSubscriber;
    
    @Override
    protected void doUpdate(final List<SelectorData> dataList) {
   
        dataList.forEach(pluginDataSubscriber::onSelectorSubscribe);
    }
}

CommonPluginDataSubscriber.onSelectorSubscribe()

CommonPluginDataSubscriber 是 PluginDataSubscriber 的唯一一个实现类：

public class CommonPluginDataSubscriber implements PluginDataSubscriber {
   
	// ...
    @Override
    public void onSelectorSubscribe(final SelectorData selectorData) {
   
        LOG.info("subscribe select data for selector: [id: {}, pluginName: {}, name: {}]", selectorData.getId(), selectorData.getPluginName(), selectorData.getName());
        subscribeDataHandler(selectorData, DataEventTypeEnum.UPDATE);
    }
    
    private <T> void subscribeDataHandler(final T classData, final DataEventTypeEnum dataType) {
   
        if (dataType == DataEventTypeEnum.UPDATE) {
   
            Optional.ofNullable(classData)
            		// 如果要更新的数据不为空，则更新缓存数据
                    .ifPresent(data -> updateCacheData(classData));
        } else if (dataType == DataEventTypeEnum.DELETE) {
   
            Optional.ofNullable(classData)
                    .ifPresent(data -> removeCacheData(classData));
        }
    }

    private <T> void updateCacheData(@NonNull final T data) {
   
        if (data instanceof PluginData) {
   
            PluginData pluginData = (PluginData) data;
            final PluginData oldPluginData = BaseDataCache.getInstance().obtainPluginData(pluginData.getName());
            BaseDataCache.getInstance().cachePluginData(pluginData);
            Optional.ofNullable(handlerMap.get(pluginData.getName()))
                    .ifPresent(handler -> handler.handlerPlugin(pluginData));

            // update enabled plugins
            PluginHandlerEventEnum state = Boolean.TRUE.equals(pluginData.getEnabled())
                    ? PluginHandlerEventEnum.ENABLED : PluginHandlerEventEnum.DISABLED;
            eventPublisher.publishEvent(new PluginHandlerEvent(state, pluginData));
            // sorted plugin
            sortPluginIfOrderChange(oldPluginData, pluginData);
            
            final String pluginName = pluginData.getName();
            // if update plugin, remove selector and rule match cache/trie cache
            if (selectorMatchConfig.getCache().getEnabled()) {
   
                MatchDataCache.getInstance().removeSelectorData(pluginName);
            }
            if (ruleMatchCacheConfig.getCache().getEnabled()) {
   
                MatchDataCache.getInstance().removeRuleData(pluginName);
            }
        } else if (data instanceof SelectorData) {
   
            SelectorData selectorData = (SelectorData) data;
            // BaseDataCache 缓存
            BaseDataCache.getInstance().cacheSelectData(selectorData);
            Optional.ofNullable(handlerMap.get(selectorData.getPluginName()))
                    .ifPresent(handler -> handler.handlerSelector(selectorData));
            // remove match cache
            if (selectorMatchConfig.getCache().getEnabled()) {
   
                MatchDataCache.getInstance().removeSelectorData(selectorData.getPluginName(), selectorData.getId());
                MatchDataCache.getInstance().removeEmptySelectorData(selectorData.getPluginName());
            }
            if (ruleMatchCacheConfig.getCache().getEnabled()) {
   
                MatchDataCache.getInstance().removeRuleDataBySelector(selectorData.getPluginName(), selectorData.getId());
                MatchDataCache.getInstance().removeEmptyRuleData(selectorData.getPluginName());
            }
            updateSelectorTrieCache(selectorData);
        } else if (data instanceof RuleData) {
   
            RuleData ruleData = (RuleData) data;
            BaseDataCache.getInstance().cacheRuleData(ruleData);
            Optional.ofNullable(handlerMap.get(ruleData.getPluginName()))
                    .ifPresent(handler -> handler.handlerRule(ruleData));
            if (ruleMatchCacheConfig.getCache().getEnabled()) {
   
                MatchDataCache.getInstance().removeRuleData(ruleData.getPluginName(), ruleData.getId());
                MatchDataCache.getInstance().removeEmptyRuleData(ruleData.getPluginName());
            }
            updateRuleTrieCache(ruleData);
        }
    }
}

3. BaseDataCache 根据数据更新缓存

网关的 SELECTOR_MAP 等缓存是由 ConcurrentMap 实现的。

筛选出不是这个 selectorId 的选择器数据，保存到 resultList 中
向 resultList 加入要更新的数据。1、2 两步相当于先删除了原 selectorId 的数据，然后再添加进新的数据
然后将更新后的 selectorData 集合排序
更新 SELECTOR_MAP

public final class BaseDataCache {
   
	// ...
   	private static final ConcurrentMap<String, List<SelectorData>> SELECTOR_MAP = Maps.newConcurrentMap();

	// 我觉得这个方法名可能是敲错了，应该是 cacheSelectorData 才对
    public void cacheSelectData(final SelectorData selectorData) {
   
        Optional.ofNullable(selectorData).ifPresent(this::selectorAccept);
    }
    
	private void selectorAccept(final SelectorData data) {
   
        String key = data.getPluginName();
        synchronized (SELECTOR_MAP) {
   
            if (SELECTOR_MAP.containsKey(key)) {
   
                // 存在 key，说明为更新操作
                List<SelectorData> existList = SELECTOR_MAP.get(key);
                // 1. 筛选出不是这个 selectorId 的选择器数据，保存到 resultList 中
                final List<SelectorData> resultList = existList.stream().filter(r -> !r.getId().equals(data.getId())).collect(Collectors.toList());
                // 2. 向 resultList 加入要更新的数据。1、2 两步相当于先删除了原 selectorId 的数据，然后再添加进新的数据
                resultList.add(data);
                // 3. 然后将更新后的 selectorData 集合排序
                final List<SelectorData> collect = resultList.stream().sorted(Comparator.comparing(SelectorData::getSort)).collect(Collectors.toList());
                // 4. 更新 SELECTOR_MAP
                SELECTOR_MAP.put(key, collect);
            } else {
   
                // 不存在 key，说明为新增操作
                SELECTOR_MAP.put(key, Lists.newArrayList(data));
            }
        }
    }
}

一张图总结

在这里插入图片描述

原文地址:https://blog.csdn.net/qq_41888672/article/details/136119333 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：https://www.suanlizi.com/kf/1758308214962786304.html 如若内容造成侵权/违法违规/事实不符，请联系《酸梨子》网邮箱：1419361763@qq.com进行投诉反馈，一经查实，立即删除！

阅读全部