sentinel框架基于SPI机制的二次开发

谁践踏了优雅 2023-10-04 08:26 12阅读 0赞

##### 概述 #####

sentinel框架是一个优秀的开源流控应用。基于sentinel框架可以实现流控降级等一系列的系统保护机制。sentinel的使用，可以参照网上的一系列教程。今天讲讲，如何基于SPI机制，扩展sentinel的功能。

sentinel虽然提供了丰富的降级限流功能，但是有时候，依旧不能满足实际的开发需要，这个时候就需要扩展sentinel框架的功能。前端页面的扩展，此处就不做过多讲解。通过实现sentinel提供的接口，可以实现规则的不同方式的持久化。

此处主要讲解server管控端扩展了新的规则后，如何在client端接收到，并应用规则。

话不多说，开始撸代码。

##### sentinel代码处理过程 #####

###### 一切的开端 ######

sentinel框架中，一切的开端，都要从一个静态类的一段静态代码块说起。

public class Env {
        
    
        public static final Sph sph = new CtSph();
    
        static {
        
            // If init fails, the process will exit.
            InitExecutor.doInit();
        }
    
    }

引入了sentinel包的应用，在启动时，初始化Env类，并执行静态方法InitExecutor.doInit()，所有的东西，都在这个初始化方法里了。

接着看下初始化方法里面做了什么。

public static void doInit() {
        
        // 只初始化一次，如果已经初始过，则直接跳出
        // 这里使用了AtomicBoolean，保证了线程安全
        if (!initialized.compareAndSet(false, true)) {
        
            return;
        }
        try {
        
            // SPI机制，获取实现了InitFunc接口的所有实现类
            ServiceLoader<InitFunc> loader = ServiceLoaderUtil.getServiceLoader(InitFunc.class);
            List<OrderWrapper> initList = new ArrayList<OrderWrapper>();
            for (InitFunc initFunc : loader) {
        
                RecordLog.info("[InitExecutor] Found init func: " + initFunc.getClass().getCanonicalName());
                // 按顺序加载实现类
                insertSorted(initList, initFunc);
            }
            // 依次调用这些实现类的init方法
            for (OrderWrapper w : initList) {
        
                w.func.init();
                RecordLog.info(String.format("[InitExecutor] Executing %s with order %d",
                                             w.func.getClass().getCanonicalName(), w.order));
            }
        } catch (Exception ex) {
        
            RecordLog.warn("[InitExecutor] WARN: Initialization failed", ex);
            ex.printStackTrace();
        } catch (Error error) {
        
            RecordLog.warn("[InitExecutor] ERROR: Initialization failed with fatal error", error);
            error.printStackTrace();
        }
    }

那么，InitFunc接口有哪些实现类呢？

有下面三个：

1，CommandCenterInitFunc.java 初始化指令中心，主要过程就是创建一个server socket来监听一个端口。server端所有的规则的增删改都会通过这个端口推送到client端。通过监听这个端口来完成server和client之间的通信。

2，MetricCallbackInit.java 这个时metric回调的钩子初始化。

3，HeartbeatSenderInitFunc.java 心跳初始化。主要处理过程就是创建一个Scheduler线程池，每10秒提交一个心跳任务，往server端注册自己的心跳。心跳包括本机的IP，CommandCenterInitFunc中监听的端口，当前应用的名称等，下面详细解释。

###### CommandCenterInitFunc ######

这个类的功能很纯粹，看代码

@InitOrder(-1)
    public class CommandCenterInitFunc implements InitFunc {
        
    
        @Override
        public void init() throws Exception {
        
            // 获取指令中心实例
            CommandCenter commandCenter = CommandCenterProvider.getCommandCenter();
    
            if (commandCenter == null) {
        
                RecordLog.warn("[CommandCenterInitFunc] Cannot resolve CommandCenter");
                return;
            }
    
            // 执行指令中心方法
            commandCenter.beforeStart();
            commandCenter.start();
            RecordLog.info("[CommandCenterInit] Starting command center: "
                    + commandCenter.getClass().getCanonicalName());
        }
    }

只做一件事，获取实例，然后执行实例方法，所有的逻辑都在实例里面了。

接下来看CommandCenterProvider里面做了什么。

public final class CommandCenterProvider {
        
    
        private static CommandCenter commandCenter = null;
    
        static {
        
            // 执行初始化方法
            resolveInstance();
        }
    
        private static void resolveInstance() {
        
            // 通过SPI获取最新的CommandCenter
            // 此处留了一个口子，如果想要重写CommandCenter，只需要创建一个新实例实现CommandCenter接口
            // 并将initOrder设置为更新即可
            CommandCenter resolveCommandCenter = SpiLoader.loadHighestPriorityInstance(CommandCenter.class);
    
            if (resolveCommandCenter == null) {
        
                RecordLog.warn("[CommandCenterProvider] WARN: No existing CommandCenter found");
            } else {
        
                // 将获取到的CommandCenter实例注册到静态变量中
                commandCenter = resolveCommandCenter;
                RecordLog.info("[CommandCenterProvider] CommandCenter resolved: " + resolveCommandCenter.getClass()
                    .getCanonicalName());
            }
        }
    
        /**
         * Get resolved {@link CommandCenter} instance.
         *
         * @return resolved {@code CommandCenter} instance
         */
        public static CommandCenter getCommandCenter() {
        
            return commandCenter;
        }
    
        private CommandCenterProvider() {
        }
    }

CommandCenterProvider中拿到了最新的CommandCenter后，执行CommandCenter中的指定方法。由于我没有扩展这个接口，所以我们直接看原生的实例：SimpleHttpCommandCenter。

直接看里面关键的两个方法

beforeStart（）

@Override
    @SuppressWarnings("rawtypes")
    public void beforeStart() throws Exception {
        
        // 依旧还是通过SPI机制，获取到所有的实现了CommandHandler接口的实例
        Map<String, CommandHandler> handlers = CommandHandlerProvider.getInstance().namedHandlers();
        // 注册这些实例到静态变量中
        registerCommands(handlers);
    }

start()

@Override
    public void start() throws Exception {
        
        int nThreads = Runtime.getRuntime().availableProcessors();
        // 初始化线程池
        this.bizExecutor = new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(10),
            new NamedThreadFactory("sentinel-command-center-service-executor"),
            new RejectedExecutionHandler() {
        
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        
                    CommandCenterLog.info("EventTask rejected");
                    throw new RejectedExecutionException();
                }
            });
    
        // 创建一个任务
        Runnable serverInitTask = new Runnable() {
        
            int port;
    
            {
        
                try {
        
                    // 获取端口号
                    port = Integer.parseInt(TransportConfig.getPort());
                } catch (Exception e) {
        
                    port = DEFAULT_PORT;
                }
            }
    
            @Override
            public void run() {
        
                boolean success = false;
                // 根据端口号，创建一个serverSocket，监听端口
                ServerSocket serverSocket = getServerSocketFromBasePort(port);
    
                if (serverSocket != null) {
        
                    CommandCenterLog.info("[CommandCenter] Begin listening at port " + serverSocket.getLocalPort());
                    socketReference = serverSocket;
                    // 提交监听任务
                    executor.submit(new ServerThread(serverSocket));
                    success = true;
                    port = serverSocket.getLocalPort();
                } else {
        
                    CommandCenterLog.info("[CommandCenter] chooses port fail, http command center will not work");
                }
    
                if (!success) {
        
                    port = PORT_UNINITIALIZED;
                }
    
                // 将端口号注册为runtimeport
                TransportConfig.setRuntimePort(port);
                executor.shutdown();
            }
    
        };
    
        new Thread(serverInitTask).start();
    }

以上就是CommandCenter的主要处理逻辑。server端除了会向client端发出同步规则(setRules)以外，还会发送其他的指令。例如同步流量信息(metric)，获取当前的规则信息（getRules）等等。这些处理过程都会被加载到Map<String, CommandHandler> handlers中，key就是命令字符串，value就是处理实例。因为这里是通过SPI来加载的，所以，这里也是一个可扩展点。

###### HeartbeatSenderInitFunc ######

@Override
    public void init() {
        
        // 获取sender实例
        HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender();
        if (sender == null) {
        
            RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");
            return;
        }
    
        // 初始化
        initSchedulerIfNeeded();
        // 获取心跳间隔
        long interval = retrieveInterval(sender);
        // 设置心跳间隔
        setIntervalIfNotExists(interval);
        // 创建心跳任务
        scheduleHeartbeatTask(sender, interval);
    }

核心的内容就是创建心跳任务。

private void scheduleHeartbeatTask(/*@NonNull*/ final HeartbeatSender sender, /*@Valid*/ long interval) {
        
        pool.scheduleAtFixedRate(new Runnable() {
        
            @Override
            public void run() {
        
                try {
        
                    // 心跳任务
                    sender.sendHeartbeat();
                } catch (Throwable e) {
        
                    RecordLog.warn("[HeartbeatSender] Send heartbeat error", e);
                }
            }
        }, 5000, interval, TimeUnit.MILLISECONDS);
        RecordLog.info("[HeartbeatSenderInit] HeartbeatSender started: "
            + sender.getClass().getCanonicalName());
    }

心跳发送的信息包括：

public class HeartbeatMessage {
        
    
        private final Map<String, String> message = new HashMap<String, String>();
    
        // 心跳包基本信息
        public HeartbeatMessage() {
        
            message.put("hostname", HostNameUtil.getHostName());
            message.put("ip", TransportConfig.getHeartbeatClientIp());
            message.put("app", AppNameUtil.getAppName());
            // Put application type (since 1.6.0).
            message.put("app_type", String.valueOf(SentinelConfig.getAppType()));
            message.put("port", String.valueOf(TransportConfig.getPort()));
        }
    
        // 扩展心跳包的入口，可以通过这个方法，扩展心跳包的内容
        public HeartbeatMessage registerInformation(String key, String value) {
        
            message.put(key, value);
            return this;
        }
    
        // 获取实际的心跳包信息
        public Map<String, String> generateCurrentMessage() {
        
            // Version of Sentinel.
            message.put("v", Constants.SENTINEL_VERSION);
            // Actually timestamp.
            message.put("version", String.valueOf(TimeUtil.currentTimeMillis()));
            message.put("port", String.valueOf(TransportConfig.getPort()));
            return message;
        }
    }

###### CommandHandler ######

上面梳理了初始化，下面继续梳理指定处理类。在SimpleHttpCommandCenter中，提交了一个监听任务ServerThread。这是一个内部类，这个类内容如下：

class ServerThread extends Thread {
        
    
        private ServerSocket serverSocket;
    
        // 构造方法
        ServerThread(ServerSocket s) {
        
            this.serverSocket = s;
            setName("sentinel-courier-server-accept-thread");
        }
    
        @Override
        public void run() {
        
            while (true) {
        
                Socket socket = null;
                try {
        
                    socket = this.serverSocket.accept();
                    setSocketSoTimeout(socket);
                    // 端口监听到数据后，包装成处理任务
                    HttpEventTask eventTask = new HttpEventTask(socket);
                    // 提交处理任务
                    bizExecutor.submit(eventTask);
                } catch (Exception e) {
        
                    ... ... 
                }
            }
        }
    }

处理任务HttpEventTask

public class HttpEventTask implements Runnable {
        
    
        ... ...
        @Override
        public void run() {
        
            ... ...
            try {
        
                ... ...
                if (firstLine.length() > 4 && StringUtil.equalsIgnoreCase("POST", firstLine.substring(0, 4))) {
        
                    // 处理post请求
                    processPostRequest(inputStream, request);
                }
    
                // 获取请求url后面的路径，作为指令名；并验证指令不为空
                String commandName = HttpCommandUtils.getTarget(request);
                if (StringUtil.isBlank(commandName)) {
        
                    writeResponse(printWriter, StatusCode.BAD_REQUEST, INVALID_COMMAND_MESSAGE);
                    return;
                }
    
                // 通过指令名称，找到对应的指令处理实例
                CommandHandler<?> commandHandler = SimpleHttpCommandCenter.getHandler(commandName);
                if (commandHandler != null) {
        
                    // 调用实例方法，处理指令
                    CommandResponse<?> response = commandHandler.handle(request);
                    // 组装响应信息
                    handleResponse(response, printWriter);
                } else {
        
                    // 未找到响应的处理实例.
                    writeResponse(printWriter, StatusCode.BAD_REQUEST, "Unknown command `" + commandName + '`');
                }
    
                long cost = System.currentTimeMillis() - start;
                CommandCenterLog.info("[SimpleHttpCommandCenter] Deal a socket task: " + firstLine
                    + ", address: " + socket.getInetAddress() + ", time cost: " + cost + " ms");
            } catch (Throwable e) {
        
                ... ...
            }
        }
    }

下面以规则更新(setRules)为例，继续说明。

setRules的实例是ModifyRulesCommandHandler。内容如下：

// CommandMapping 设置名称和描述
    @CommandMapping(name = "setRules", desc = "modify the rules, accept param: type={ruleType}&data={ruleJson}")
    public class ModifyRulesCommandHandler implements CommandHandler<String> {
        
        private static final int FASTJSON_MINIMAL_VER = 0x01020C00;
    
        // 处理方法
        @Override
        public CommandResponse<String> handle(CommandRequest request) {
        
            ... ...
            // 规则类型
            String type = request.getParam("type");
            // 规则数据
            String data = request.getParam("data");
            if (StringUtil.isNotEmpty(data)) {
        
                try {
        
                    // 解码
                    data = URLDecoder.decode(data, "utf-8");
                } catch (Exception e) {
        
                    RecordLog.info("Decode rule data error", e);
                    return CommandResponse.ofFailure(e, "decode rule data error");
                }
            }
    
            RecordLog.info("Receiving rule change (type: {}): {}", type, data);
    
            String result = "success";
    
            // 根据类型，调用不同的规则处理
            // 流控规则
            if (FLOW_RULE_TYPE.equalsIgnoreCase(type)) {
        
                List<FlowRule> flowRules = JSONArray.parseArray(data, FlowRule.class);
                FlowRuleManager.loadRules(flowRules);
                if (!writeToDataSource(getFlowDataSource(), flowRules)) {
        
                    result = WRITE_DS_FAILURE_MSG;
                }
                return CommandResponse.ofSuccess(result);
            } else if (AUTHORITY_RULE_TYPE.equalsIgnoreCase(type)) {
         // 黑白名单规则
                List<AuthorityRule> rules = JSONArray.parseArray(data, AuthorityRule.class);
                AuthorityRuleManager.loadRules(rules);
                if (!writeToDataSource(getAuthorityDataSource(), rules)) {
        
                    result = WRITE_DS_FAILURE_MSG;
                }
                return CommandResponse.ofSuccess(result);
            } else if (DEGRADE_RULE_TYPE.equalsIgnoreCase(type)) {
         // 降级规则
                List<DegradeRule> rules = JSONArray.parseArray(data, DegradeRule.class);
                DegradeRuleManager.loadRules(rules);
                if (!writeToDataSource(getDegradeDataSource(), rules)) {
        
                    result = WRITE_DS_FAILURE_MSG;
                }
                return CommandResponse.ofSuccess(result);
            } else if (SYSTEM_RULE_TYPE.equalsIgnoreCase(type)) {
         // 系统规则
                List<SystemRule> rules = JSONArray.parseArray(data, SystemRule.class);
                SystemRuleManager.loadRules(rules);
                if (!writeToDataSource(getSystemSource(), rules)) {
        
                    result = WRITE_DS_FAILURE_MSG;
                }
                return CommandResponse.ofSuccess(result);
            }
            return CommandResponse.ofFailure(new IllegalArgumentException("invalid type"));
        }
    }

以流控为例，接收到新的规则后，调用FlowRuleManager来处理新规则。

public static void loadRules(List<FlowRule> rules) {
        
        // 调用update方法
        currentProperty.updateValue(rules);
    }

currentProperty.updateValue回调监听类的configUpdate方法。至此，规则完成了从server端到client端的同步。

private static final class FlowPropertyListener implements PropertyListener<List<FlowRule>> {
        
    
        @Override
        public void configUpdate(List<FlowRule> value) {
        
            Map<String, List<FlowRule>> rules = FlowRuleUtil.buildFlowRuleMap(value);
            if (rules != null) {
        
                flowRules.clear();
                flowRules.putAll(rules);
            }
            RecordLog.info("[FlowRuleManager] Flow rules received: " + flowRules);
        }
    
        @Override
        public void configLoad(List<FlowRule> conf) {
        
            Map<String, List<FlowRule>> rules = FlowRuleUtil.buildFlowRuleMap(conf);
            if (rules != null) {
        
                flowRules.clear();
                flowRules.putAll(rules);
            }
            RecordLog.info("[FlowRuleManager] Flow rules loaded: " + flowRules);
        }
    }

以上就是sentinel框架的规则同步流程。

##### 如何扩展 #####

接下来，才是重要内容。如何扩展sentinel框架？其实在了解了整个的处理过程之后，扩展点也就迎刃而解。基本上，凡是用到了SPI机制的地方，都可以作为扩展的入口。

###### InitFunc接口 ######

这个接口是初始化的入口。这里可以用到的扩展点是，基于zk等注册中心的扩展。原生的sentinel框架里面，使用的是CommandCenterInitFunc来接收并处理规则更新。实际的生产应用中，可以使用zk作为注册中心，然后实现一个基于zk的InitFunc实例，这个实例用于创建zk监听，监听zk上的指定路径。通过监听路径下的规则值变化，主动pull规则数据到client端。实际上，sentinel框架已经集成了zk，通过引入zk相关的jar包，就可以使用这部分代码。

<dependency>
      <groupId>com.alibaba.csp</groupId>
      <artifactId>sentinel-datasource-zookeeper</artifactId>
    </dependency>
    <dependency>
      <groupId>org.apache.curator</groupId>
      <artifactId>curator-recipes</artifactId>
      <version>${curator.version}</version>
      <scope>compile</scope>
    </dependency>

除此之外，扩展InitFunc接口，还可以用于新增其他的一些初始化配置。

###### CommandCenter接口 ######

扩展这个接口，可以让sentinel使用新的指令中心。如上述所说，使用zk做注册中心。对于zk注册中心规则变更的逻辑，也可以通过重写一个高版本的指令中心用于从zk获取规则信息。这样的好处在于，不会多出一套规则获取的机制。使用InitFunc做zk扩展，原生的ServerSocket机制依然生效，并监听接口。如果使用CommandCenter接口扩展，则只会有一个zk机制来同步规则数据，socket监听机制将不会生效。

原因在于获取CommandCenter时，取的是最新的CommandCenter。

CommandCenter resolveCommandCenter = SpiLoader.loadHighestPriorityInstance(CommandCenter.class);

###### CommandHandler接口 ######

用于扩展指令。如果需要在sentinel框架中新增指令，那么可以通过扩展这个接口来添加新的指令。具体实现参照上述处理过程。

例如，实现一个扩展的规则同步指令：setExtendRules

@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) {
        
        	// todo 具体的处理逻辑，可参照原生的指令处理器ModifyRulesCommandHandler
        }
    }

实现完伤处接口后，在server端，发送指令setExtendRules就可以调用这边的逻辑进行后续处理。

###### ProcessorSlot接口 ######

这里额外补充一个接口。扩展这个接口，可以扩展sentinel框架在处理资源时的处理链。当调用这个静态方法的时候，会触发一系列的处理链。其中包括限流，降级等。如果需要扩展这个处理链，则可以扩展这个接口。

entry = SphU.entry(resource);