🚀 新增清理过期message pipe定时线程

Author: hengboy

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/MessagePipe.java

@@ -162,7 +162,6 @@ public synchronized void handleFirst(Function<Message, MessageProcessStatus> fun
         log.debug("The message pipe：{} scheduler thread is woken up, handing first message.", name);
         Message current = null;
         MessageProcessStatus status = MessageProcessStatus.SEND_SUCCESS;
-        Long currentTimeMillis = System.currentTimeMillis();
         RLock takeLock = redissonClient.getLock(takeLockName);
         try {
             MessagePipeConfiguration.LockTime lockTime = configuration.getLockTime();
@@ -182,7 +181,7 @@ public synchronized void handleFirst(Function<Message, MessageProcessStatus> fun
         } catch (Exception e) {
             this.doHandleException(e, status, current);
         } finally {
-            lastProcessTimeMillis.set(currentTimeMillis);
+            lastProcessTimeMillis.set(System.currentTimeMillis());
             transfer = true;
             if (takeLock.isLocked() && takeLock.isHeldByCurrentThread()) {
                 takeLock.unlock();
@@ -218,6 +217,7 @@ public synchronized void handleToLast(Function<Message, MessageProcessStatus> fu
         } catch (Exception e) {
             this.doHandleException(e, status, current);
         } finally {
+            lastProcessTimeMillis.set(System.currentTimeMillis());
             transfer = true;
             runningHandleAll = false;
             if (takeLock.isLocked() && takeLock.isHeldByCurrentThread()) {

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/config/ServerConfiguration.java

@@ -38,6 +38,14 @@ public class ServerConfiguration {
      * The default number of core threads for monitoring and scheduling thread pools
      */
     private int coreThreadPoolSize = 20;
+    /**
+     * Interval for cleaning up expired message pipe threads, in seconds
+     */
+    private long cleanupExpiredMessagePipeIntervalSeconds = 10;
+    /**
+     * The threshold for determining an expired message pipe, in seconds
+     */
+    private long cleanupExpiredMessagePipeThresholdSeconds = 1800;
     /**
      * Configure the message pipe name to exclude distribution
      * <p>

message-pipe-server/src/main/java/org/minbox/framework/message/pipe/server/manager/AbstractMessagePipeManager.java

@@ -14,13 +14,13 @@
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.ObjectUtils;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.concurrent.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 /**
  * The {@link MessagePipeManager} abstract implementation class
@@ -36,13 +36,15 @@ public abstract class AbstractMessagePipeManager implements MessagePipeManager,
      * The Key of the Map is the name of the {@link MessagePipe}
      */
     private static final ConcurrentMap<String, MessagePipe> MESSAGE_PIPE_MAP = new ConcurrentHashMap();
+    private static final int CLEANUP_EXPIRED_CORE_THREADS = 1;
     /**
      * Create the configuration object used by the {@link MessagePipe}
      */
     private MessagePipeConfiguration sharedConfiguration;
     private BeanFactory beanFactory;
     private static ExecutorService SCHEDULER_SERVICE;
     private static ExecutorService MONITOR_SERVICE;
+    private static ScheduledExecutorService CLEANUP_EXPIRED_SERVICE;
     private ServerConfiguration serverConfiguration;
     private MessagePipeFactoryBean messagePipeFactoryBean;
     private ServiceDiscovery serviceDiscovery;
@@ -146,10 +148,33 @@ public void afterPropertiesSet() throws Exception {
         this.serviceDiscovery = beanFactory.getBean(ServiceDiscovery.class);
         SCHEDULER_SERVICE = Executors.newFixedThreadPool(serverConfiguration.getCoreThreadPoolSize());
         MONITOR_SERVICE = Executors.newFixedThreadPool(serverConfiguration.getCoreThreadPoolSize());
+        CLEANUP_EXPIRED_SERVICE = Executors.newScheduledThreadPool(CLEANUP_EXPIRED_CORE_THREADS);
+        this.startCleanupExpiredThread();
         log.info("The MessagePipeManager startup successfully，maximum number of message pipes：{}.",
                 serverConfiguration.getMaxMessagePipeCount());
     }
 
+    /**
+     * Start cleanup expired message pipe thread
+     */
+    private void startCleanupExpiredThread() {
+        CLEANUP_EXPIRED_SERVICE.scheduleAtFixedRate(() -> {
+            log.debug("Clean up expired message pipes thread is start working...");
+            long checkTimeMillis = System.currentTimeMillis();
+            List<MessagePipe> expiredList = MESSAGE_PIPE_MAP.values().stream()
+                    .filter(messagePipe -> {
+                        long diffSeconds = TimeUnit.MILLISECONDS.toSeconds(checkTimeMillis - messagePipe.getLastProcessTimeMillis());
+                        return diffSeconds > serverConfiguration.getExpiredExcludeThresholdSeconds();
+                    }).collect(Collectors.toList());
+            if (!ObjectUtils.isEmpty(expiredList)) {
+                expiredList.stream().forEach(expiredMessagePipe ->
+                        MESSAGE_PIPE_MAP.remove(expiredMessagePipe.getName(), expiredMessagePipe));
+                log.warn("The cleanup of expired message pipes thread is completed, this cleanup: {}.",
+                        expiredList.stream().map(MessagePipe::getName).collect(Collectors.toList()));
+            }
+        }, 1, serverConfiguration.getCleanupExpiredMessagePipeIntervalSeconds(), TimeUnit.SECONDS);
+    }
+
     @Override
     public void destroy() throws Exception {
         SCHEDULER_SERVICE.shutdown();