sentinel指令扩展实践

小咪咪 2023-10-04 08:28 12阅读 0赞

如果需要扩展sentinel的指令集，需要从何入手？怎么样扩展才会显得优雅？

下面介绍一个我在工作中的实践经验，供大家参考。

以重写降级逻辑为例，我们重新定义降级逻辑控制台的增删查改逻辑就不做过多介绍，我们从扩展的降级规则发送到client端开始说起。

#### server端发送规则 ####

首先，我们可以仿照sentinel原生的规则发送方法，写一个扩展的规则发送方法

private boolean setExtendRules(String app, String ip, int port, String type, List<? extends RuleEntity> entities) {
        
        if (entities == null) {
        
            return true;
        }
        try {
        
            AssertUtil.notEmpty(app, "Bad app name");
            AssertUtil.notEmpty(ip, "Bad machine IP");
            AssertUtil.isTrue(port > 0, "Bad machine port");
            String data = JSON.toJSONString(
                    entities.stream().map(RuleEntity::toRule).collect(Collectors.toList()));
            data = URLEncoder.encode(data, "utf-8") ;
            Map<String, String> params = new HashMap<>(2);
            params.put("type", type);
            params.put("data", data);
            String result = executeCommand(app, ip, port, SET_EXTEND_RULES_PATH, params, true).get();
            logger.info("setExtendRules result: {}, type={}", result, type);
            return true;
        } catch (InterruptedException e) {
        
            logger.warn("setExtendRules API failed: {}", type, e);
            return false;
        } catch (ExecutionException e) {
        
            logger.warn("setExtendRules API failed: {}", type, e.getCause());
            return false;
        } catch (Exception e) {
        
            logger.error("setExtendRules API failed, type={}", type, e);
            return false;
        }
    }

其中，扩展命令为：

private static final String SET_EXTEND_RULES_PATH = "setExtendRules";

入参type为指令类型，如果扩展多个指令，可通过这个type来进行分发。

这样我们就在server端定义了一个setExtendRules命令的发送API。当扩展的规则有增删改时，就可以调用这个API去推送规则到client端。到此，server端的功能就说这么多，接下来client端是重点。

#### client端接收规则 ####

##### 命令监听器 #####

首先，我们需要扩展一个命令处理器，server端发送过来的命令被serverSocket监听到之后，会调用命令监听器SimpleHttpCommandCenter，并根据CommandMapping定义的内容来调用特定的CommandHandler。所以，我们需要扩展一个CommandHandler来处理我们新增的指令。

核心代码如下

@CommandMapping(name = "setExtendRules", desc = "modify the extend rules, accept param: type={ruleType}&data={ruleJson}")
    public class ModifyExtendRulesCommandHandler implements CommandHandler<String> {
        
        private static final int FASTJSON_MINIMAL_VER = 0x01020C00;
    
        @Override
        public CommandResponse<String> handle(CommandRequest request) {
        
            // XXX from 1.7.2, force to fail when fastjson is older than 1.2.12
            // We may need a better solution on this.
            if (VersionUtil.fromVersionString(JSON.VERSION) < FASTJSON_MINIMAL_VER) {
        
                // fastjson too old
                return CommandResponse.ofFailure(new RuntimeException("The \"fastjson-" + JSON.VERSION
                        + "\" introduced in application is too old, you need fastjson-1.2.12 at least."));
            }
            String type = request.getParam("type");
            // rule data in get parameter
            String data = request.getParam("data");
            if (StringUtil.isNotEmpty(data)) {
        
                try {
        
                    data = URLDecoder.decode(data, "utf-8");
                } catch (Exception e) {
        
                    logger.info("Decode rule data error", e);
                    return CommandResponse.ofFailure(e, "decode rule data error");
                }
            }
    
            logger.info("Receiving rule change (type: {}): {}", type, data);
    
            String result = "success";
            
            // 根据type来指定处理类型
            if( EXTEND_RULE_TYPE.equalsIgnoreCase(type)){
        
                List<ExtendRule> rules = JSONArray.parseArray(data, ExtendRule.class);
                // 加载规则
                ExtendRuleManager.loadRules(rules);
                return CommandResponse.ofSuccess(result);
            }
            return CommandResponse.ofFailure(new IllegalArgumentException("invalid type"));
        }
    
        private static final String WRITE_DS_FAILURE_MSG = "partial success (write data source failed)";
        private static final String EXTEND_RULE_TYPE = "extendRule";
    }

##### 规则管理器 #####

在上面的代码中，有一个ExtendRuleManager，这个就是一个规则管理器。新接收到的规则，都会通过ExtendRuleManager.loadRules方法加载到内存中去。因此，我们也需要扩展一个规则管理器。核心代码如下

public final class ExtendRuleManager {
        
    
        // 内存中存储的规则map
        private static final Map<String, Set<ExtendRule>> ExtendRules = new ConcurrentHashMap<>();
    
        // 规则监听器，监听规则变化
        private static final RulePropertyListener LISTENER = new RulePropertyListener();
        // 内存中当前存储的规则列表
        private static SentinelProperty<List<ExtendRule>> currentProperty
            = new DynamicSentinelProperty<>();
    
        static {
        
            // 添加监听器
            currentProperty.addListener(LISTENER);
        }
    
        /**
         * 监听SentinelProperty中的ExtendRule规则列表。其中property就是源规则ExtendRules，
         * 可通过 loadRules(List) 方法直接设置规则列表
         */
        public static void register2Property(SentinelProperty<List<ExtendRule>> property) {
        
            AssertUtil.notNull(property, "property cannot be null");
            synchronized (LISTENER) {
        
                logger.info("[ExtendRuleManager] Registering new property to extend rule manager");
                currentProperty.removeListener(LISTENER);
                property.addListener(LISTENER);
                currentProperty = property;
            }
        }
    
        /**
        * 校验规则
        */
        public static void checkExtendRule(ResourceWrapper resource, Context context, DefaultNode node, int count)
            throws BlockException {
        
    
            // 根据资源名称获取规则列表
            Set<ExtendRule> rules = ExtendRules.get(resource.getName());
            if (rules == null) {
        
                return;
            }
    
            // 校验所有的规则是否通过，不通过，则抛出指定异常
            for (ExtendRule rule : rules) {
        
                if (!rule.passCheck(context, node, count)) {
        
                    throw new ExtendRuleException(rule.getLimitApp(), rule);
                }
            }
        }
    
    
        /**
         * 加载规则列表
         */
        public static void loadRules(List<ExtendRule> rules) {
        
            try {
        
                // 调用更新，会触发规则监听器的configUpdate方法
                currentProperty.updateValue(rules);
            } catch (Throwable e) {
        
                logger.warn("[ExtendRuleManager] Unexpected error when loading extend rules", e);
            }
        }
    
        /**
        * 内部类，规则监听器，
        */
        private static class RulePropertyListener implements PropertyListener<List<ExtendRule>> {
        
    
            /**
            * 规则更新方法
            * 设置为清空全部，重新加载，和load方法一致
            */
            @Override
            public void configUpdate(List<ExtendRule> conf) {
        
                // 从config中获取规则map，方法略，可参照原有manager实现
                Map<String, Set<ExtendRule>> rules = loadExtendConf(conf);
                if (rules != null) {
        
                    ExtendRules.clear();
                    ExtendRules.putAll(rules);
                }
                logger.info("[ExtendRuleManager] extend rules received: " + ExtendRules);
            }
    
            /**
            * 规则加载方法
            * 设置为清空全部，重新加载
            */
            @Override
            public void configLoad(List<ExtendRule> conf) {
        
                Map<String, Set<ExtendRule>> rules = loadExtendConf(conf);
                if (rules != null) {
        
                    ExtendRules.clear();
                    ExtendRules.putAll(rules);
                }
                logger.info("[ExtendRuleManager] extend rules loaded: " + ExtendRules);
            }
        }
    }

##### 规则处理器 #####

规则load到client端的manager中，并存储到内存里。那么什么时候会调用到呢？sentinel所有的规则调用，都是一个入口

Entry entry = SphU.entry(resource);

当需要降级限流的时候，只要调用这个方法去创建一个entry，就会触发一个处理链调用。SphU.entry方法的核心代码如下

// 获取调用链
    ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
    
    /*
     * 如果没有调用链，则直接返回一个新的entry
     */
    if (chain == null) {
        
        return new CtEntry(resourceWrapper, null, context);
    }
    
    // 创建带调用链的Entry
    Entry e = new CtEntry(resourceWrapper, chain, context);
    try {
        
        // 启动链条
        chain.entry(context, resourceWrapper, null, count, prioritized, args);
    } catch (BlockException e1) {
        
        e.exit(count, args);
        throw e1;
    } catch (Throwable e1) {
        
        // This should not happen, unless there are errors existing in Sentinel internal.
        RecordLog.info("Sentinel unexpected exception", e1);
    }
    return e;

链条获取，先是通过SPI获取一个链条的实例，如果二次开发比较深的话，这里也是一个扩展点，可以重写链条的实例。我们这里就不考虑这么深，直接使用链条的实例。获取到链条实例后，调用链条的build方法，创建调用链。

public class DefaultSlotChainBuilder implements SlotChainBuilder {
        
    
        @Override
        public ProcessorSlotChain build() {
        
            ProcessorSlotChain chain = new DefaultProcessorSlotChain();
    
            // 通过SPI获取调用链节点。入口接口为ProcessorSlot
            // 注：所有ProcessorSlot的实例都必须不同，因为这些调用链并非无状态的。
            List<ProcessorSlot> sortedSlotList = SpiLoader.loadPrototypeInstanceListSorted(ProcessorSlot.class);
            for (ProcessorSlot slot : sortedSlotList) {
        
                // 所有调用链都必须继承一个调用链的适配器
                if (!(slot instanceof AbstractLinkedProcessorSlot)) {
        
                    RecordLog.warn("The ProcessorSlot(" + slot.getClass().getCanonicalName() + ") is not an instance of AbstractLinkedProcessorSlot, can't be added into ProcessorSlotChain");
                    continue;
                }
    
                chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
            }
    
            return chain;
        }
    }

从上面可以知道，所有的规则处理，都是通过这个调用链的节点来完成的。这个设计，类似DataSource中的filter插件的设计思想。因此，我们只要扩展一个自己的ProcessorSlot，就可以让框架去执行我们扩展的规则。但是处理器不直接继承ProcessorSlot，而是继承一个适配器AbstractLinkedProcessorSlot，代码示例如下：

@SpiOrder(-1000)
    public class ExtendRuleSlot extends AbstractLinkedProcessorSlot<DefaultNode> {
        
    
        /**
        * entry创建时的调用链
        */
        @Override
        public void entry(Context context, ResourceWrapper resourceWrapper, DefaultNode node, int count, boolean prioritized, Object... args)
            throws Throwable {
        
            // 调用规则管理器，校验规则
            ExtendRuleManager.checkExtendRule(resourceWrapper, context, node, count);
            // 继续调用链中的后续节点调用
            fireEntry(context, resourceWrapper, node, count, prioritized, args);
        }
    
        /**
        * entry退出时的调用链
        */
        @Override
        public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
        
            fireExit(context, resourceWrapper, count, args);
        }
    }

#### 总结 ####

以上，就是关于扩展sentinel规则的实践全流程。上面没有提到规则的实际校验逻辑，这个根据自己的业务来定。这个规则是直接写在ruleEntity里面的，里面适配了规则校验的实际逻辑。

整个流程使用sentinel提供的扩展接口接入sentinel框架。sentinel框架使用了大量的SPI机制。以上说明中，关于SPI的扩展，均需要在项目的resources目录下创建路径META-INF.services，并在路径下创建配置文件。  
![在这里插入图片描述][20210316152847911.png]  
文件名就是实现的接口的全路径名，文件内容就是扩展的实现类的全路径名。多个实现类可以多行输入。

[20210316152847911.png]: https://img-blog.csdnimg.cn/20210316152847911.png