split into channel-server-starter and channel-server-app modules

2024-04-21 10:21:39 +00:00 · 2020-09-08 17:50:15 +08:00
parent 7d7b685c55
commit a705ba97a0
32 changed files with 445 additions and 153 deletions
@@ -0,0 +1,76 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>arthas-all</artifactId>
+        <groupId>com.taobao.arthas</groupId>
+        <version>3.3.10-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>arthas-channel-server-app</artifactId>
+    <name>arthas-channel-server-app</name>
+
+    <properties>
+        <maven.compiler.target>1.8</maven.compiler.target>
+        <maven.compiler.source>1.8</maven.compiler.source>
+    </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <!-- Import dependency management from Spring Boot -->
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>${spring-boot.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.taobao.arthas</groupId>
+            <artifactId>arthas-channel-server-starter</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <showDeprecation>true</showDeprecation>
+                </configuration>
+            </plugin>
+
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>2.1.7.RELEASE</version>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
@@ -4,7 +4,6 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.scheduling.annotation.EnableScheduling;

-@EnableScheduling
@SpringBootApplication
 public class ArthasChannelApplication {

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>arthas-all</artifactId>
+        <groupId>com.taobao.arthas</groupId>
+        <version>3.3.10-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>arthas-channel-server-starter</artifactId>
+    <name>arthas-channel-server-starter</name>
+
+    <properties>
+        <maven.compiler.target>1.8</maven.compiler.target>
+        <maven.compiler.source>1.8</maven.compiler.source>
+    </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <!-- Import dependency management from Spring Boot -->
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>${spring-boot.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>com.taobao.arthas</groupId>
+            <artifactId>arthas-channel-server</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>
@@ -1,5 +1,6 @@
-package com.alibaba.arthas.channel.server.configuration;
+package com.alibaba.arthas.channel.server.autoconfigure;

+import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
 import com.alibaba.arthas.channel.server.grpc.ArthasServiceGrpcImpl;
 import com.alibaba.arthas.channel.server.grpc.ChannelServer;
 import com.alibaba.arthas.channel.server.message.MessageExchangeService;
@@ -7,13 +8,15 @@ import com.alibaba.arthas.channel.server.message.impl.MessageExchangeServiceImpl
 import com.alibaba.arthas.channel.server.redis.RedisAgentManageServiceImpl;
 import com.alibaba.arthas.channel.server.redis.RedisMessageExchangeServiceImpl;
 import com.alibaba.arthas.channel.server.service.AgentBizSerivce;
+import com.alibaba.arthas.channel.server.service.AgentCleaner;
 import com.alibaba.arthas.channel.server.service.AgentManageService;
 import com.alibaba.arthas.channel.server.service.ApiActionDelegateService;
 import com.alibaba.arthas.channel.server.service.impl.AgentBizServiceImpl;
 import com.alibaba.arthas.channel.server.service.impl.AgentManageServiceImpl;
 import com.alibaba.arthas.channel.server.service.impl.ApiActionDelegateServiceImpl;
 import com.alibaba.arthas.channel.server.ws.WebSocketServer;
-import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -29,16 +32,16 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;

-/**
- * @author gongdewei 2020/8/14
- */
+
@Configuration
+@ConditionalOnClass(ChannelServer.class)
@EnableConfigurationProperties(ChannelServerProperties.class)
-public class ChannelServerConfiguration {
+public class ChannelServerAutoConfiguration {

    @Bean
-    public ScheduledExecutorService executorService() {
-        // 设置较大的corePoolSize，避免长时间运行的task阻塞调度队列 (https://developer.aliyun.com/article/5897 "1.2 线程数量控制")
+    @ConditionalOnMissingBean
+    public ScheduledExecutorConfig scheduledExecutorConfig() {
+        // 设置较大的corePoolSize，避免并发运行的task阻塞调度队列 (https://developer.aliyun.com/article/5897 "1.2 线程数量控制")
        int corePoolSize = 10;
        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(corePoolSize, new ThreadFactory() {
            @Override
@@ -49,12 +52,15 @@ public class ChannelServerConfiguration {
            }
        });

-        //ScheduledThreadPoolExecutor为无界队列，设置MaximumPoolSize无效
-//        if (executorService instanceof ThreadPoolExecutor) {
-//            ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorService;
-//            threadPoolExecutor.setMaximumPoolSize(50);
-//        }
-        return executorService;
+        ScheduledExecutorConfig scheduledExecutorConfig = new ScheduledExecutorConfig();
+        scheduledExecutorConfig.setExecutorService(executorService);
+        return scheduledExecutorConfig;
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public AgentBizSerivce agentBizSerivce() {
+        return new AgentBizServiceImpl();
    }

    @Bean
@@ -62,25 +68,33 @@ public class ChannelServerConfiguration {
        return new ApiActionDelegateServiceImpl();
    }

-    @Bean
-    public AgentBizSerivce agentBizSerivce() {
-        return new AgentBizServiceImpl();
-    }
-
    @Bean
    public ArthasServiceGrpcImpl arthasServiceGrpc() {
        return new ArthasServiceGrpcImpl();
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
+    public AgentCleaner agentCleaner(ScheduledExecutorConfig scheduledExecutorConfig, ChannelServerProperties serverProperties) {
+        ChannelServerProperties.Agent agentConfig = serverProperties.getAgent();
+        AgentCleaner agentCleaner = new AgentCleaner(scheduledExecutorConfig);
+        agentCleaner.setCleanIntervalMills(agentConfig.getCleanIntervalMills());
+        agentCleaner.setRemovingTimeout(agentConfig.getRemovingTimeoutMills());
+        agentCleaner.setDownTimeout(agentConfig.getDownTimeoutMills());
+        agentCleaner.setOutOfServiceTimeout(agentConfig.getOutOfServiceTimeoutMills());
+        return agentCleaner;
+    }
+
+    @Bean(initMethod = "start", destroyMethod = "stop")
+    @ConditionalOnMissingBean
    @ConditionalOnProperty(value = "channel.server.backend.enabled", havingValue = "true", matchIfMissing = false)
-    public ChannelServer channelServer(@Value("${channel.server.backend.port}") int port) {
+    public ChannelServer channelServer(ChannelServerProperties serverProperties) {
        ChannelServer channelServer = new ChannelServer();
-        channelServer.setPort(port);
+        channelServer.setPort(serverProperties.getBackend().getPort());
        return channelServer;
    }

    @Bean(initMethod = "start", destroyMethod = "stop")
+    @ConditionalOnMissingBean
    @ConditionalOnProperty(value = "channel.server.websocket.enabled", havingValue = "true", matchIfMissing = false)
    public WebSocketServer webSocketServer(ChannelServerProperties serverProperties) {
        WebSocketServer server = new WebSocketServer();
@@ -96,11 +110,13 @@ public class ChannelServerConfiguration {
    static class StandaloneConfiguration {

        @Bean
+        @ConditionalOnMissingBean
        public AgentManageService agentManageService() {
            return new AgentManageServiceImpl();
        }

        @Bean
+        @ConditionalOnMissingBean
        public MessageExchangeService messageExchangeService() {
            return new MessageExchangeServiceImpl();
        }
@@ -111,16 +127,19 @@ public class ChannelServerConfiguration {
    @Configuration
    static class RedisConfiguration {
        @Bean
+        @ConditionalOnMissingBean
        public AgentManageService agentManageService() {
            return new RedisAgentManageServiceImpl();
        }

        @Bean
+        @ConditionalOnMissingBean
        public MessageExchangeService messageExchangeService() {
            return new RedisMessageExchangeServiceImpl();
        }

        @Bean
+        @ConditionalOnMissingBean
        public ReactiveRedisTemplate<String, byte[]> reactiveRedisTemplate(ReactiveRedisConnectionFactory redisConnectionFactory) {
            ReactiveRedisTemplate<String, byte[]> template = new ReactiveRedisTemplate (redisConnectionFactory, RedisSerializationContext
                    .<String, byte[]>newSerializationContext()
@@ -133,6 +152,7 @@ public class ChannelServerConfiguration {
        }

        @Bean
+        @ConditionalOnMissingBean
        public ReactiveStringRedisTemplate reactiveStringRedisTemplate(ReactiveRedisConnectionFactory redisConnectionFactory) {
            ReactiveStringRedisTemplate template = new ReactiveStringRedisTemplate(redisConnectionFactory);
            return template;
@@ -0,0 +1,115 @@
+package com.alibaba.arthas.channel.server.autoconfigure;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "channel.server")
+public class ChannelServerProperties {
+
+    private Server websocket = new Server();
+    private Server backend = new Server();
+
+    private Agent agent = new Agent();
+
+    public Server getWebsocket() {
+        return websocket;
+    }
+
+    public void setWebsocket(Server websocket) {
+        this.websocket = websocket;
+    }
+
+    public Server getBackend() {
+        return backend;
+    }
+
+    public void setBackend(Server backend) {
+        this.backend = backend;
+    }
+
+    public Agent getAgent() {
+        return agent;
+    }
+
+    public void setAgent(Agent agent) {
+        this.agent = agent;
+    }
+
+    public static class Server {
+        private String host;
+        private int port;
+        private boolean ssl;
+        private boolean enabled;
+
+        public String getHost() {
+            return host;
+        }
+
+        public void setHost(String host) {
+            this.host = host;
+        }
+
+        public int getPort() {
+            return port;
+        }
+
+        public void setPort(int port) {
+            this.port = port;
+        }
+
+        public boolean isSsl() {
+            return ssl;
+        }
+
+        public void setSsl(boolean ssl) {
+            this.ssl = ssl;
+        }
+
+        public boolean isEnabled() {
+            return enabled;
+        }
+
+        public void setEnabled(boolean enabled) {
+            this.enabled = enabled;
+        }
+    }
+
+    public static class Agent {
+        private int cleanIntervalMills = 5000;
+        private int outOfServiceTimeoutMills = 15*1000;
+        private int downTimeoutMills = 30*1000;
+        private int removingTimeoutMills = 600*1000;
+
+        public int getCleanIntervalMills() {
+            return cleanIntervalMills;
+        }
+
+        public void setCleanIntervalMills(int cleanIntervalMills) {
+            this.cleanIntervalMills = cleanIntervalMills;
+        }
+
+        public int getOutOfServiceTimeoutMills() {
+            return outOfServiceTimeoutMills;
+        }
+
+        public void setOutOfServiceTimeoutMills(int outOfServiceTimeoutMills) {
+            this.outOfServiceTimeoutMills = outOfServiceTimeoutMills;
+        }
+
+        public int getDownTimeoutMills() {
+            return downTimeoutMills;
+        }
+
+        public void setDownTimeoutMills(int downTimeoutMills) {
+            this.downTimeoutMills = downTimeoutMills;
+        }
+
+        public int getRemovingTimeoutMills() {
+            return removingTimeoutMills;
+        }
+
+        public void setRemovingTimeoutMills(int removingTimeoutMills) {
+            this.removingTimeoutMills = removingTimeoutMills;
+        }
+    }
+
+}
@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.alibaba.arthas.channel.server.autoconfigure.ChannelServerAutoConfiguration
@@ -102,20 +102,6 @@
                    <showDeprecation>true</showDeprecation>
                </configuration>
            </plugin>
-
-            <plugin>
-                <groupId>org.springframework.boot</groupId>
-                <artifactId>spring-boot-maven-plugin</artifactId>
-                <version>2.1.7.RELEASE</version>
-                <executions>
-                    <execution>
-                        <phase>package</phase>
-                        <goals>
-                            <goal>repackage</goal>
-                        </goals>
-                    </execution>
-                </executions>
-            </plugin>
        </plugins>
    </build>

@@ -0,0 +1,20 @@
+package com.alibaba.arthas.channel.server.conf;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+/**
+ * ScheduledExecutorService config for channel server, avoiding bean injecting conflicts
+ * @author gongdewei 2020/9/8
+ */
+public class ScheduledExecutorConfig {
+
+    private ScheduledExecutorService executorService;
+
+    public ScheduledExecutorService getExecutorService() {
+        return executorService;
+    }
+
+    public void setExecutorService(ScheduledExecutorService executorService) {
+        this.executorService = executorService;
+    }
+}
@@ -1,66 +0,0 @@
-package com.alibaba.arthas.channel.server.configuration;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-
-@ConfigurationProperties(prefix = "channel.server")
-public class ChannelServerProperties {
-
-    private Server websocket;
-    private Server backend;
-
-    public Server getWebsocket() {
-        return websocket;
-    }
-
-    public void setWebsocket(Server websocket) {
-        this.websocket = websocket;
-    }
-
-    public Server getBackend() {
-        return backend;
-    }
-
-    public void setBackend(Server backend) {
-        this.backend = backend;
-    }
-
-    public static class Server {
-        private String host;
-        private int port;
-        private boolean ssl;
-        private boolean enabled = true;
-
-        public String getHost() {
-            return host;
-        }
-
-        public void setHost(String host) {
-            this.host = host;
-        }
-
-        public int getPort() {
-            return port;
-        }
-
-        public void setPort(int port) {
-            this.port = port;
-        }
-
-        public boolean isSsl() {
-            return ssl;
-        }
-
-        public void setSsl(boolean ssl) {
-            this.ssl = ssl;
-        }
-
-        public boolean isEnabled() {
-            return enabled;
-        }
-
-        public void setEnabled(boolean enabled) {
-            this.enabled = enabled;
-        }
-    }
-
-}
@@ -1,5 +1,6 @@
 package com.alibaba.arthas.channel.server.grpc;

+import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 import org.slf4j.Logger;
@@ -20,14 +21,14 @@ public class ChannelServer {
    private ArthasServiceGrpcImpl arthasServiceGrpc;

    @Autowired
-    private ScheduledExecutorService executorService;
+    private ScheduledExecutorConfig executorServiceConfig;

    private Server server;
    private int port = 7700;

    public void start() throws Exception  {

-        executorService.submit(new Runnable() {
+        executorServiceConfig.getExecutorService().submit(new Runnable() {
            @Override
            public void run() {
                try {
@@ -1,5 +1,6 @@
 package com.alibaba.arthas.channel.server.message.impl;

+import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
 import com.alibaba.arthas.channel.server.message.MessageExchangeException;
 import com.alibaba.arthas.channel.server.message.MessageExchangeService;
 import com.alibaba.arthas.channel.server.message.topic.Topic;
@@ -14,7 +15,6 @@ import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;

 /**
@@ -28,7 +28,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService {
    private Map<Topic, TopicData> topicMap = new ConcurrentHashMap<Topic, TopicData>();

    @Autowired
-    private ScheduledExecutorService executorService;
+    private ScheduledExecutorConfig executorServiceConfig;

    @Override
    public void createTopic(Topic topic) throws MessageExchangeException {
@@ -108,7 +108,7 @@ public class MessageExchangeServiceImpl implements MessageExchangeService {
        topicData.setTimeout(timeout);

        final TopicData finalTopicData = topicData;
-        executorService.submit(new Runnable() {
+        executorServiceConfig.getExecutorService().submit(new Runnable() {
            @Override
            public void run() {
                while (true) {
@@ -1,5 +1,6 @@
 package com.alibaba.arthas.channel.server.redis;

+import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
 import com.alibaba.arthas.channel.server.message.MessageExchangeException;
 import com.alibaba.arthas.channel.server.message.MessageExchangeService;
 import com.alibaba.arthas.channel.server.message.topic.Topic;
@@ -13,7 +14,6 @@ import reactor.core.publisher.Mono;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ScheduledExecutorService;

 /**
 * Message exchange for cluster channel server
@@ -30,7 +30,7 @@ public class RedisMessageExchangeServiceImpl implements MessageExchangeService {
    private ReactiveRedisTemplate<String, byte[]> redisTemplate;

    @Autowired
-    private ScheduledExecutorService executorService;
+    private ScheduledExecutorConfig executorServiceConfig;

    @Override
    public void createTopic(Topic topic) throws MessageExchangeException {
@@ -82,7 +82,7 @@ public class RedisMessageExchangeServiceImpl implements MessageExchangeService {
        Mono<byte[]> mono = redisTemplate.opsForList().rightPop(topic.getTopic(), Duration.ofMillis(timeout));
        mono.doOnSuccess(messageBytes -> {
            //schedule running, avoid blocking redis reactive
-            executorService.submit(() -> {
+            executorServiceConfig.getExecutorService().submit(() -> {
                if (messageBytes != null) {
                    boolean next = messageHandler.onMessage(messageBytes);
                    if (next) {
@@ -96,7 +96,7 @@ public class RedisMessageExchangeServiceImpl implements MessageExchangeService {
                }
            });
        }).doOnError(throwable -> {
-            executorService.submit(() -> {
+            executorServiceConfig.getExecutorService().submit(() -> {
                if (throwable instanceof QueryTimeoutException) {
                    //ignore Redis command timed out
                    subscribe(topic, timeout, messageHandler);
@@ -9,7 +9,5 @@ public interface AgentBizSerivce {

    void heartbeat(String agentId, String agentStatus, String agentVersion);

-    void cleanOutdatedAgents();
-
    void compareAndUpdateAgentStatus(String agentId, AgentStatus expectedStatus, AgentStatus newStatus);
 }
@@ -0,0 +1,124 @@
+package com.alibaba.arthas.channel.server.service;
+
+import com.alibaba.arthas.channel.proto.AgentStatus;
+import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
+import com.alibaba.arthas.channel.server.model.AgentVO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+public class AgentCleaner {
+
+    private static final Logger logger = LoggerFactory.getLogger(AgentCleaner.class);
+    private int removingTimeout = 60000;
+    private int downTimeout = 30000;
+    private int outOfServiceTimeout = 15000;
+    private int cleanIntervalMills = 5000;
+
+    @Autowired
+    private AgentManageService agentManageService;
+    private ScheduledExecutorConfig scheduledExecutorConfig;
+    private ScheduledFuture<?> scheduledFuture;
+
+    public AgentCleaner(ScheduledExecutorConfig scheduledExecutorConfig) {
+        this.scheduledExecutorConfig = scheduledExecutorConfig;
+    }
+
+    public void start() {
+
+        if (downTimeout < outOfServiceTimeout) {
+            throw new IllegalArgumentException(" agent down timeout must not be less than out of service timeout");
+        }
+
+        if (removingTimeout > 0 && removingTimeout < downTimeout) {
+            throw new IllegalArgumentException(" agent removing timeout must not be less than down timeout or -1");
+        }
+
+        scheduledFuture = scheduledExecutorConfig.getExecutorService().scheduleWithFixedDelay(new Runnable() {
+            @Override
+            public void run() {
+                doClean();
+            }
+        }, cleanIntervalMills, cleanIntervalMills, TimeUnit.MILLISECONDS);
+    }
+
+    public void stop() {
+        if (scheduledFuture != null) {
+            scheduledFuture.cancel(true);
+        }
+    }
+
+    public void doClean() {
+        long now = System.currentTimeMillis();
+        Mono<List<AgentVO>> agentsMono = agentManageService.listAgents();
+        agentsMono.doOnSuccess(agents -> {
+            for (AgentVO agent : agents) {
+                long heartbeatDelay = now - agent.getHeartbeatTime();
+                if (heartbeatDelay > removingTimeout && removingTimeout > 0) {
+                    logger.info("clean up dead agent: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay);
+                    agentManageService.removeAgentById(agent.getAgentId());
+                } else if (heartbeatDelay > downTimeout) {
+                    if (!AgentStatus.DOWN.name().equals(agent.getAgentStatus())) {
+                        logger.info("Mark agent status as DOWN, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay);
+                        agent.setAgentStatus(AgentStatus.DOWN.name());
+                        agentManageService.updateAgent(agent);
+                    }
+                } else if (heartbeatDelay > outOfServiceTimeout) {
+                    if (!AgentStatus.OUT_OF_SERVICE.name().equals(agent.getAgentStatus())) {
+                        logger.info("Mark agent status as OUT_OF_SERVICE, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay);
+                        agent.setAgentStatus(AgentStatus.OUT_OF_SERVICE.name());
+                        agentManageService.updateAgent(agent);
+                    }
+                }
+            }
+        }).doOnError(throwable -> {
+            logger.error("clean agent error", throwable);
+        }).subscribe();
+    }
+
+    public int getRemovingTimeout() {
+        return removingTimeout;
+    }
+
+    public void setRemovingTimeout(int removingTimeout) {
+        this.removingTimeout = removingTimeout;
+    }
+
+    public int getDownTimeout() {
+        return downTimeout;
+    }
+
+    public void setDownTimeout(int downTimeout) {
+        if (downTimeout < 5000) {
+            throw new IllegalArgumentException("agent down timeout must be not be less than 5000 mills");
+        }
+        this.downTimeout = downTimeout;
+    }
+
+    public int getOutOfServiceTimeout() {
+        return outOfServiceTimeout;
+    }
+
+    public void setOutOfServiceTimeout(int outOfServiceTimeout) {
+        if (outOfServiceTimeout < 5000) {
+            throw new IllegalArgumentException("agent out of service timeout must not be less than 5000 mills");
+        }
+        this.outOfServiceTimeout = outOfServiceTimeout;
+    }
+
+    public int getCleanIntervalMills() {
+        return cleanIntervalMills;
+    }
+
+    public void setCleanIntervalMills(int cleanIntervalMills) {
+        if (cleanIntervalMills < 2000) {
+            throw new IllegalArgumentException("agent clean interval must not be less than 2000 mills");
+        }
+        this.cleanIntervalMills = cleanIntervalMills;
+    }
+}
@@ -8,7 +8,6 @@ import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.scheduling.annotation.Scheduled;
 import reactor.core.publisher.Mono;

 import java.util.List;
@@ -43,38 +42,6 @@ public class AgentBizServiceImpl implements AgentBizSerivce {
        }).subscribe();
    }

-    @Override
-    @Scheduled(fixedDelayString = "5000")
-    public void cleanOutdatedAgents() {
-        long now = System.currentTimeMillis();
-        Mono<List<AgentVO>> agentsMono = agentManageService.listAgents();
-        agentsMono.doOnSuccess(new Consumer<List<AgentVO>>() {
-            @Override
-            public void accept(List<AgentVO> agents) {
-                for (AgentVO agent : agents) {
-                    long heartbeatDelay = now - agent.getHeartbeatTime();
-                    if (heartbeatDelay > 60000) {
-                        logger.info("clean up dead agent: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay);
-                        agentManageService.removeAgentById(agent.getAgentId());
-                    } else if (heartbeatDelay > 30000) {
-                        if (!AgentStatus.DOWN.name().equals(agent.getAgentStatus())) {
-                            logger.info("Mark agent status as DOWN, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay);
-                            agent.setAgentStatus(AgentStatus.DOWN.name());
-                            agentManageService.updateAgent(agent);
-                        }
-                    } else if (heartbeatDelay > 15000) {
-                        if (!AgentStatus.OUT_OF_SERVICE.name().equals(agent.getAgentStatus())) {
-                            logger.info("Mark agent status as OUT_OF_SERVICE, agentId: {}, heartbeat delay: {}", agent.getAgentId(), heartbeatDelay);
-                            agent.setAgentStatus(AgentStatus.OUT_OF_SERVICE.name());
-                            agentManageService.updateAgent(agent);
-                        }
-                    }
-                }
-            }
-        }).subscribe();
-
-    }
-
    @Override
    public void compareAndUpdateAgentStatus(String agentId, AgentStatus expectedStatus, AgentStatus newStatus) {
        agentManageService.findAgentById(agentId).flatMap(optionalAgentVO -> {
@@ -4,6 +4,7 @@ import com.alibaba.arthas.channel.proto.ActionResponse;
 import com.alibaba.arthas.channel.proto.ExecuteResult;
 import com.alibaba.arthas.channel.proto.ResponseStatus;
 import com.alibaba.arthas.channel.server.api.ApiException;
+import com.alibaba.arthas.channel.server.conf.ScheduledExecutorConfig;
 import com.alibaba.arthas.channel.server.model.AgentVO;
 import com.alibaba.arthas.channel.server.api.ApiRequest;
 import com.alibaba.arthas.channel.server.api.ApiResponse;
@@ -24,7 +25,6 @@ import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
-import org.springframework.web.client.HttpClientErrorException;
 import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 import reactor.core.publisher.Mono;
@@ -32,7 +32,6 @@ import reactor.core.publisher.Mono;
 import java.io.IOException;
 import java.util.List;
 import java.util.Optional;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;

 /**
@@ -50,7 +49,7 @@ public class AgentController {
    private ApiActionDelegateService apiActionDelegateService;

    @Autowired
-    private ScheduledExecutorService executorService;
+    private ScheduledExecutorConfig executorServiceConfig;


    @RequestMapping("/agents")
@@ -174,7 +173,7 @@ public class AgentController {
                                          @RequestBody final String requestBody,
                                          @RequestParam(value = "timeout", defaultValue = "300000") final int timeout) {
        final SseEmitter emitter = new SseEmitter(Long.valueOf(timeout));
-        executorService.submit(new Runnable() {
+        executorServiceConfig.getExecutorService().submit(new Runnable() {
            public void run() {
                ApiResponse apiResponse;
                try {
@@ -29,9 +29,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Optional;

-/**
- * @author gongdewei 2020/9/2
- */
+
 public class WebSocketFrameHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketFrameHandler.class);
@@ -74,6 +74,8 @@
        <module>channel/channel-common</module>
        <module>channel/channel-server</module>
        <module>channel/channel-client</module>
+        <module>channel/channel-server-starter</module>
+        <module>channel/channel-server-app</module>
    </modules>

    <profiles>