sentinel源码解析之-sentinel dashboard与客户端连接建立过程

素颜马尾好姑娘i 2022-12-15 03:36 303阅读 0赞

因为项目用的是spring cloud，对于服务熔断一直用的是默认的hystrix

hystrix是一个比较强大的开源框架，支持线程池隔离和信号量隔离。但是很遗憾，正如github上hystrix描述的那样：

> Hystrix is no longer in active development, and is currently in maintenance mode.

是的，hystrix目前已经停止开发好几年了

从这个框架的发展历程也不禁让我觉得每个软件项目都是具有生命的，从青涩到壮年，弹指间匆匆流年，hystrix就是一个很好的典范

尽管hystrix已经停止开发了，但它的设计理念与设计思想具有着很深远的意义，这在后面出现的resilience4j和sentinel中都得到了体现

*向hystrix致敬*  
![https://github.com/Netflix/Hystrix][https_github.com_Netflix_Hystrix]  
Hystrix的github官方地址为：  
[https://github.com/Netflix/Hystrix][https_github.com_Netflix_Hystrix 1]

--------------------

因为项目的需要，为了观察统计方便，需要实时查看每个服务的接口的QPS数据，加上hystrix已停止开发的原因综合考虑后我决定将现有服务中的hystrix用sentinel来进行替换

在将服务接入sentinel的过程中刚开始也遇到了些问题，虽然后面都进行了解决，为了再次理解这里还是决定将部分问题和思考在这里进行一个简短的记录  
![https://github.com/alibaba/Sentinel][https_github.com_alibaba_Sentinel]  
sentinel的github官方地址为：  
[https://github.com/alibaba/Sentinel][https_github.com_alibaba_Sentinel 1]

**sentinel dashboard与客户端连接建立过程**  
之所以要记录这个是因为我将sentinel dashboard配置好了后，也将客户端启动好了配置感觉也没有问题，但sentinel dashboard上依旧看不到任何的服务列表，为此感觉了解sentinel dashboard与被监控的服务的连接过程很有必要，只有通过源码分析才能更好的了解具体的连接过程是怎么样的

## spring mvc项目所需依赖包 ##

如果项目中spring boot的项目，其引入sentinel dashboard的代码在sentinel的demo中也有示例。

加入如下的依赖

<dependency>
                <groupId>com.alibaba.csp</groupId>
                <artifactId>sentinel-core</artifactId>
            </dependency>
            <dependency>
                <groupId>com.alibaba.csp</groupId>
                <artifactId>sentinel-transport-simple-http</artifactId>
            </dependency>
            <dependency>
                <groupId>com.alibaba.csp</groupId>
                <artifactId>sentinel-spring-webmvc-adapter</artifactId>
                <version>${project.version}</version>
            </dependency>

其中sentinel-core是sentinel的核心代码，sentinel-spring-webmvc-adapter是为了适配spring mvc项目的适配器,sentinel-transport-simple-http是与sentinel dashboard建立连接的客户端。

## SentinelWebInterceptor主流程 ##

之后再在spring mvc的项目中添加一个servlet的拦截器就可以了,也就是SentinelWebInterceptor，在spring mvc项目中对于sentinel源码分析入口也就是这个拦截器

当访问某一个接口时，因为有拦截器的原因会进入SentinelWebInterceptor的preHandle方法，也就是它父类com.alibaba.csp.sentinel.adapter.spring.webmvc.AbstractSentinelInterceptor的preHandle方法，此方法源码如下：

@Override
        public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
            throws Exception {
            try {
                String resourceName = getResourceName(request);
    
                if (StringUtil.isEmpty(resourceName)) {
                    return true;
                }
                
                if (increaseReferece(request, this.baseWebMvcConfig.getRequestRefName(), 1) != 1) {
                    return true;
                }
                
                // Parse the request origin using registered origin parser.
                String origin = parseOrigin(request);
                String contextName = getContextName(request);
                ContextUtil.enter(contextName, origin);
                Entry entry = SphU.entry(resourceName, ResourceTypeConstants.COMMON_WEB, EntryType.IN);
                request.setAttribute(baseWebMvcConfig.getRequestAttributeName(), entry);
                return true;
            } catch (BlockException e) {
                try {
                    handleBlockException(request, response, e);
                } finally {
                    ContextUtil.exit();
                }
                return false;
            }
        }

其中会执行这一行：  
Entry entry = SphU.entry(resourceName, ResourceTypeConstants.COMMON\_WEB, EntryType.IN);

/**
         * Record statistics and perform rule checking for the given resource.
         *
         * @param name         the unique name for the protected resource
         * @param resourceType classification of the resource (e.g. Web or RPC)
         * @param trafficType  the traffic type (inbound, outbound or internal). This is used
         *                     to mark whether it can be blocked when the system is unstable,
         *                     only inbound traffic could be blocked by {@link SystemRule}
         * @return the {@link Entry} of this invocation (used for mark the invocation complete and get context data)
         * @throws BlockException if the block criteria is met (e.g. metric exceeded the threshold of any rules)
         * @since 1.7.0
         */
        public static Entry entry(String name, int resourceType, EntryType trafficType) throws BlockException {
            return Env.sph.entryWithType(name, resourceType, trafficType, 1, OBJECTS0);
        }

当执行Env.sph下的方法，会将Env类激活，Env类的代码如下：

/**
     * Sentinel Env. This class will trigger all initialization for Sentinel.
     *
     * <p>
     * NOTE: to prevent deadlocks, other classes' static code block or static field should
     * NEVER refer to this class.
     * </p>
     *
     * @author jialiang.linjl
     */
    public class Env {
    
        public static final Sph sph = new CtSph();
    
        static {
            // If init fails, the process will exit.
            InitExecutor.doInit();
        }
    
    }

## spi加载InitFunc类 ##

当Env激活后，会执行static代码块,执行InitExecutor.doInit()方法，其源码如下：

/**
     * Load registered init functions and execute in order.
     *
     * @author Eric Zhao
     */
    public final class InitExecutor {
    
        private static AtomicBoolean initialized = new AtomicBoolean(false);
    
        /**
         * If one {@link InitFunc} throws an exception, the init process
         * will immediately be interrupted and the application will exit.
         *
         * The initialization will be executed only once.
         */
        public static void doInit() {
            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);
                }
                for (OrderWrapper w : initList) {
                    //遍历进行初始化
                    w.func.init();
                    RecordLog.info("[InitExecutor] Executing {} with order {}",
                        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();
            }
        }
        ……
    }

从源码可以看出InitExecutor的doInit方法会通过spi的方式找到InitFunc的实现类，并执行初始化

此项目中因为导入了sentinel-core和sentinel-transport-simple-http包，则通过上面的spi加载InitFunc.class则会加载出如下类(从resoures/META-INF.services下加载)：

*  com.alibaba.csp.sentinel.transport.init.CommandCenterInitFunc
 *  com.alibaba.csp.sentinel.transport.init.HeartbeatSenderInitFunc
 *  com.alibaba.csp.sentinel.metric.extension.MetricCallbackInit

前两个类是从sentinel-transport-simple-http的parent项目sentinel-transport-common中加载出来的，最后一个MetricCallbackInit是从sentinel-core中加载的

然后代码会依次执行上面三个InitFunc的init()方法

为了走马观花，这里只看sentinel-transport-simple-http中的两个方法

## CommandCenterInitFunc ##

其init方法代码为：

@Override
        public void init() throws Exception {
            //通过spi加载出对应的commandCenter
            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.getCommandCenter()也是通过spi根据项目需要加载出对应的CommandCenter,要么是com.alibaba.csp.sentinel.transport.command.NettyHttpCommandCenter要么就是com.alibaba.csp.sentinel.transport.command.SimpleHttpCommandCenter，这里因为前面引用的是sentinel-transport-simple-http包的原因，则最终拿到的会是SimpleHttpCommandCenter

## SimpleHttpCommandCenter ##

主要看下这个commandCenter的beforeStart()和start()方法

### SimpleHttpCommandCenter.beforeStart源码 ###

public void beforeStart() throws Exception {
            // Register handlers
            Map<String, CommandHandler> handlers = CommandHandlerProvider.getInstance().namedHandlers();
            registerCommands(handlers);
        }

上面的方法还是通过spi加载出项目中的CommandHandler.class类，然后放到一个handlerMap中

### SimpleHttpCommandCenter.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 = 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;
                    }
    
                    TransportConfig.setRuntimePort(port);
                    executor.shutdown();
                }
    
            };
    
            new Thread(serverInitTask).start();
        }

上面的serverInitTask初始化了一个httpserver,其server的代码为上面中的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) {
                        CommandCenterLog.info("Server error", e);
                        if (socket != null) {
                            try {
                                socket.close();
                            } catch (Exception e1) {
                                CommandCenterLog.info("Error when closing an opened socket", e1);
                            }
                        }
                        try {
                            // In case of infinite log.
                            Thread.sleep(10);
                        } catch (InterruptedException e1) {
                            // Indicates the task should stop.
                            break;
                        }
                    }
                }
            }
        }

原本以为sentinel至少会使用nio来实现httpserver，但看到这里却发现sentinel的客户端是也用传统的bio+线程池来实现的，阻塞监听请求，来一个就将它用HttpEventTask封装下提交到bizExecutor中

再简单看下HttpEventTask这个类,看下run和writeResponse方法就行了,

@Override
        public void run() {
            if (socket == null) {
                return;
            }
    
            PrintWriter printWriter = null;
            InputStream inputStream = null;
            try {
                long start = System.currentTimeMillis();
                inputStream = new BufferedInputStream(socket.getInputStream());
                OutputStream outputStream = socket.getOutputStream();
    
                printWriter = new PrintWriter(
                    new OutputStreamWriter(outputStream, Charset.forName(SentinelConfig.charset())));
    
                String firstLine = readLine(inputStream);
                CommandCenterLog.info("[SimpleHttpCommandCenter] Socket income: " + firstLine
                    + ", addr: " + socket.getInetAddress());
                CommandRequest request = processQueryString(firstLine);
    
                if (firstLine.length() > 4 && StringUtil.equalsIgnoreCase("POST", firstLine.substring(0, 4))) {
                    // Deal with post method
                    processPostRequest(inputStream, request);
                }
    
                // Validate the target command.
                String commandName = HttpCommandUtils.getTarget(request);
                if (StringUtil.isBlank(commandName)) {
                    writeResponse(printWriter, StatusCode.BAD_REQUEST, INVALID_COMMAND_MESSAGE);
                    return;
                }
    
                // Find the matching command handler.
                CommandHandler<?> commandHandler = SimpleHttpCommandCenter.getHandler(commandName);
                if (commandHandler != null) {
                    CommandResponse<?> response = commandHandler.handle(request);
                    handleResponse(response, printWriter);
                } else {
                    // No matching command handler.
                    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 (RequestException e) {
                writeResponse(printWriter, e.getStatusCode(), e.getMessage());
            } catch (Throwable e) {
                CommandCenterLog.warn("[SimpleHttpCommandCenter] CommandCenter error", e);
                try {
                    if (printWriter != null) {
                        String errorMessage = SERVER_ERROR_MESSAGE;
                        e.printStackTrace();
                        if (!writtenHead) {
                            writeResponse(printWriter, StatusCode.INTERNAL_SERVER_ERROR, errorMessage);
                        } else {
                            printWriter.println(errorMessage);
                        }
                        printWriter.flush();
                    }
                } catch (Exception e1) {
                    CommandCenterLog.warn("Failed to write error response", e1);
                }
            } finally {
                closeResource(inputStream);
                closeResource(printWriter);
                closeResource(socket);
            }
        }
        
        private void writeResponse(PrintWriter out, StatusCode statusCode, String message) {
            out.print("HTTP/1.0 " + statusCode.toString() + "\r\n"
                + "Content-Length: " + (message == null ? 0 : message.getBytes().length) + "\r\n"
                + "Connection: close\r\n\r\n");
            if (message != null) {
                out.print(message);
            }
            out.flush();
            writtenHead = true;
        }
        
        ……

发现了啥，这简直就是一个手写http服务器教程的感觉

看到这里其实就对sentinel客户端是如何接收处理来自sentinel dashboard的请求有了一个大致的了解

## HeartbeatSenderInitFunc ##

在看了接收sentinel dashboard的请求源码后，继续来看下发送数据给sentinel dashboard的源码,直接从HeartbeatSenderInitFunc来看：

import java.util.concurrent.ScheduledExecutorService;
    import java.util.concurrent.ScheduledThreadPoolExecutor;
    import java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy;
    import java.util.concurrent.TimeUnit;
    
    import com.alibaba.csp.sentinel.concurrent.NamedThreadFactory;
    import com.alibaba.csp.sentinel.config.SentinelConfig;
    import com.alibaba.csp.sentinel.heartbeat.HeartbeatSenderProvider;
    import com.alibaba.csp.sentinel.init.InitFunc;
    import com.alibaba.csp.sentinel.init.InitOrder;
    import com.alibaba.csp.sentinel.log.RecordLog;
    import com.alibaba.csp.sentinel.transport.HeartbeatSender;
    import com.alibaba.csp.sentinel.transport.config.TransportConfig;
    
    /**
     * Global init function for heartbeat sender.
     *
     * @author Eric Zhao
     */
    @InitOrder(-1)
    public class HeartbeatSenderInitFunc implements InitFunc {
    
        private ScheduledExecutorService pool = null;
    
        private void initSchedulerIfNeeded() {
            if (pool == null) {
                pool = new ScheduledThreadPoolExecutor(2,
                    new NamedThreadFactory("sentinel-heartbeat-send-task", true),
                    new DiscardOldestPolicy());
            }
        }
    
        @Override
        public void init() {
            //通过spi加载出对应的HeartbeatSender
            HeartbeatSender sender = HeartbeatSenderProvider.getHeartbeatSender();
            if (sender == null) {
                RecordLog.warn("[HeartbeatSenderInitFunc] WARN: No HeartbeatSender loaded");
                return;
            }
            //初始化一个线程池
            initSchedulerIfNeeded();
            long interval = retrieveInterval(sender);
            //设置心跳数据发送间隔时间
            setIntervalIfNotExists(interval);
            //定义sender的执行时间
            scheduleHeartbeatTask(sender, interval);
        }
    
        private boolean isValidHeartbeatInterval(Long interval) {
            return interval != null && interval > 0;
        }
    
        private void setIntervalIfNotExists(long interval) {
            SentinelConfig.setConfig(TransportConfig.HEARTBEAT_INTERVAL_MS, String.valueOf(interval));
        }
    
        long retrieveInterval(/*@NonNull*/ HeartbeatSender sender) {
            Long intervalInConfig = TransportConfig.getHeartbeatIntervalMs();
            if (isValidHeartbeatInterval(intervalInConfig)) {
                RecordLog.info("[HeartbeatSenderInitFunc] Using heartbeat interval "
                    + "in Sentinel config property: " + intervalInConfig);
                return intervalInConfig;
            } else {
                long senderInterval = sender.intervalMs();
                RecordLog.info("[HeartbeatSenderInit] Heartbeat interval not configured in "
                    + "config property or invalid, using sender default: " + senderInterval);
                return senderInterval;
            }
        }
    
        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());
        }
    }

前面已经提过，sentinel在初始化时会执行InitExecutor.doInit()，在其中会通过spi加载出项目中的所有InitFunc的实现类，然后再依次执行每个实现类的init方法,因为HeartbeatSenderInitFunc是也是一个InitFunc的实现类，所以直接看它的init方法

通过查看HeartbeatSenderInitFunc的init方法我们可以看到它仍旧是通过spi加载出对应的HeartbeatSender实现类，在spring mvc的项目中则会加载出com.alibaba.csp.sentinel.transport.heartbeat.SimpleHttpHeartbeatSender这个类,然后再执行scheduleHeartbeatTask方法给定义调用sendHeartbeat()方法的时间间隔

其心跳数据发送时间由变量interval来确定,其值由retrieveInterval方法来获取，从其retrieveInterval方法的源码我们可以看出其值会先通过csp.sentinel.heartbeat.interval.ms获取，如果没有配置，则会用默认值DEFAULT\_INTERVAL的值，也就是10秒

> private static final long DEFAULT\_INTERVAL = 1000 \* 10;

通过观察线程池的配置代码我们可以发现，如果我们配置的心跳数据发送间隔太短，线程池还没来得及处理，则会丢弃以前的老任务

pool = new ScheduledThreadPoolExecutor(2,
                    new NamedThreadFactory("sentinel-heartbeat-send-task", true),
                    new DiscardOldestPolicy());

### SimpleHttpHeartbeatSender ###

再看下SimpleHttpHeartbeatSender的心跳数据发送方法,直接看它的sendHeartheat方法：

@Override
        public boolean sendHeartbeat() throws Exception {
            if (TransportConfig.getRuntimePort() <= 0) {
                RecordLog.info("[SimpleHttpHeartbeatSender] Command server port not initialized, won't send heartbeat");
                return false;
            }
            //获取出dashboard的地址和端口
            Tuple2<String, Integer> addrInfo = getAvailableAddress();
            if (addrInfo == null) {
                return false;
            }
    
            InetSocketAddress addr = new InetSocketAddress(addrInfo.r1, addrInfo.r2);
            SimpleHttpRequest request = new SimpleHttpRequest(addr, TransportConfig.getHeartbeatApiPath());
            request.setParams(heartBeat.generateCurrentMessage());
            try {
                SimpleHttpResponse response = httpClient.post(request);
                if (response.getStatusCode() == OK_STATUS) {
                    return true;
                } else if (clientErrorCode(response.getStatusCode()) || serverErrorCode(response.getStatusCode())) {
                    RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr
                        + ", http status code: " + response.getStatusCode());
                }
            } catch (Exception e) {
                RecordLog.warn("[SimpleHttpHeartbeatSender] Failed to send heartbeat to " + addr, e);
            }
            return false;
        }

上面这段代码看着还算是比较简单的，会先执行getAvailableAddress方法获取出dashboard的地址和端口，然后再执行TransportConfig.getHeartbeatApiPath()获取出dashboard接收心跳数据的请求的地址，最后万事俱备后直接post过去就行了

所以如果我们发现sentinel的客户端总是连接不上dashboard或dashboard总是接收不到sentinel客户端的数据，那么我们就可以通过跟踪代码的方式来查看所对应的dashboard的ip和端口是否正确即可。

通过查看getAvailableAddress方法我们可以定位到系统中加载的其实就是csp.sentinel.dashboard.server这个配置的值，以ip:port的方式配置即可

> String config = SentinelConfig.getConfig(CONSOLE\_SERVER);

> public static final String CONSOLE\_SERVER = “csp.sentinel.dashboard.server”;

如果没有配置此配置，则会执行直接返回false,不进行发送

最后来个项目改造后的sentinel dashboard截图  
![sentinel dashborad效果图][sentinel dashborad]

## 总结 ##

在简单看了sentinel dashboard的源码后个人感觉这个框架的源码还是比较好理解的，目前来看此项目中用了大量的spi，sentinel dashboard与客户端传输的方式是用的bio+线程池来实现的，其他的还需要慢慢研究

[https_github.com_Netflix_Hystrix]: /images/20221123/368e597723814c9fa16db99ad5799b79.png
[https_github.com_Netflix_Hystrix 1]: https://github.com/Netflix/Hystrix
[https_github.com_alibaba_Sentinel]: /images/20221123/9608d1727581427aad1d8698b5f21b97.png
[https_github.com_alibaba_Sentinel 1]: https://github.com/alibaba/Sentinel
[sentinel dashborad]: /images/20221123/22b731cd958f402d8ce281ee2251969d.png