feat: 修复线程池泄露问题，并增加相应测试。

Author: magicliang

trpc-container/trpc-container-default/src/test/java/com/tencent/trpc/container/container/ConsumerProxyTest.java

@@ -22,6 +22,7 @@
 import com.tencent.trpc.core.exception.TRpcException;
 import com.tencent.trpc.core.rpc.RpcClientContext;
 import com.tencent.trpc.core.rpc.TRpcProxy;
+import com.tencent.trpc.proto.support.DefResponseFutureManager;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -44,6 +45,7 @@ public void after() {
             container.stop();
         }
         ConfigManager.stopTest();
+        DefResponseFutureManager.reset();
     }
 
     @Test

trpc-core/src/main/java/com/tencent/trpc/core/common/ConfigManager.java

@@ -44,6 +44,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.WeakHashMap;
 import java.util.concurrent.TimeUnit;
 
 public class ConfigManager {
@@ -55,6 +56,12 @@ public class ConfigManager {
      * Container startup listener
      */
     private final Set<TRPCRunListener> tRPCRunListeners = Sets.newConcurrentHashSet();
+
+    /**
+     * Shutdown listeners for decoupled cleanup, using WeakHashMap to prevent classloader leaks
+     */
+    private final Map<ShutdownListener, Boolean> shutdownListeners = new WeakHashMap<>();
+
     /**
      * Business initialization
      */
@@ -107,6 +114,7 @@ public static void stopTest() {
         RpcClusterClientManager.reset();
         RpcServerManager.reset();
         WorkerPoolManager.reset();
+
         instance.setDefault = false;
     }
 
@@ -251,6 +259,41 @@ public void addTRPCRunListener(TRPCRunListener trpcRunListener) {
         tRPCRunListeners.add(trpcRunListener);
     }
 
+    /**
+     * Get all shutdown listeners loaded via SPI.
+     * 
+     * @return set of shutdown listeners
+     */
+    public void registerShutdownListener(ShutdownListener listener) {
+        if (listener != null) {
+            shutdownListeners.put(listener, Boolean.TRUE);
+        }
+    }
+
+    /**
+     * Unregister a shutdown listener.
+     *
+     * @param listener the shutdown listener to unregister
+     */
+    public void unregisterShutdownListener(ShutdownListener listener) {
+        if (listener != null) {
+            shutdownListeners.remove(listener);
+        }
+    }
+
+    /**
+     * Notify all registered shutdown listeners.
+     */
+    private void notifyShutdownListeners() {
+        shutdownListeners.keySet().forEach(listener -> {
+            try {
+                listener.onShutdown();
+            } catch (Exception e) {
+                logger.error("Error executing shutdown listener: {}", listener.getClass().getName(), e);
+            }
+        });
+    }
+
     @Override
     public String toString() {
         return "ApplicationConfig {globalConfig=" + globalConfig + ", serverConfig=" + serverConfig
@@ -301,10 +344,10 @@ private void warmupSelector(BackendConfig backendConfig) {
                         try {
                             String selectorId = nm.getSelectorId();
                             SelectorManager.getManager().get(selectorId).warmup(backendConfig.toNamingServiceId());
-                            logger.warn("Warm up selector success。(selectorId={},naming={}) ", selectorId,
+                            logger.warn("Warm up selector success.(selectorId={},naming={}) ", selectorId,
                                     backendConfig.getNamingOptions().getServiceNaming());
                         } catch (Exception ex) {
-                            logger.warn("Warm up selector exception。(selectorId={}, naming={}) ",
+                            logger.warn("Warm up selector exception.(selectorId={}, naming={}) ",
                                     nm.getSelectorId(), nm.getServiceNaming(), ex);
                         }
                     });
@@ -385,6 +428,10 @@ protected void stopInternal() throws Exception {
             ExtensionLoader.destroyAllPlugin();
             // 8) close client cluster
             RpcClusterClientManager.close();
+
+            // Notify all shutdown listeners before actual shutdown
+            notifyShutdownListeners();
+
             logger.info(">>>tRPC Server stopped");
         }
 
@@ -394,4 +441,4 @@ public String toString() {
         }
     }
 
-}
+}

trpc-core/src/main/java/com/tencent/trpc/core/common/ShutdownListener.java

@@ -0,0 +1,26 @@
+/*
+ * Tencent is pleased to support the open source community by making tRPC available.
+ *
+ * Copyright (C) 2023 Tencent.
+ * All rights reserved.
+ *
+ * If you have downloaded a copy of the tRPC source code from Tencent,
+ * please note that tRPC source code is licensed under the Apache 2.0 License,
+ * A copy of the Apache 2.0 License can be found in the LICENSE file.
+ */
+
+package com.tencent.trpc.core.common;
+
+/**
+ * Shutdown listener for components that need to perform cleanup when the container stops.
+ * This provides a decoupled way for components to register shutdown hooks without
+ * creating circular dependencies.
+ */
+public interface ShutdownListener {
+    
+    /**
+     * Called when the container is shutting down.
+     */
+    void onShutdown();
+
+}

trpc-core/src/main/java/com/tencent/trpc/core/management/support/MBeanRegistryHelper.java

@@ -36,4 +36,20 @@ public static void registerMBean(Object object, ObjectName objectName) {
         }
     }
 
+    /**
+     * Unregister mbean
+     *
+     * @param objectName mbean object name {@link ObjectName}
+     */
+    public static void unregisterMBean(ObjectName objectName) {
+        try {
+            MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
+            if (mBeanServer.isRegistered(objectName)) {
+                mBeanServer.unregisterMBean(objectName);
+            }
+        } catch (Exception e) {
+            logger.warn("unregister mbean exception: ", e);
+        }
+    }
+
 }

trpc-core/src/main/java/com/tencent/trpc/core/rpc/def/DefTimeoutManager.java

@@ -34,6 +34,7 @@ public DefTimeoutManager(int tickms) {
     public Future<?> watch(final Runnable task, long timeout) {
         FutureAdapter<?> adapter = new FutureAdapter(task);
         adapter.wrap = timer.newTimeout(adapter, timeout, TimeUnit.MILLISECONDS);
+
         return adapter;
     }
 

trpc-core/src/main/java/com/tencent/trpc/core/worker/support/thread/ForkJoinWorkerPool.java

@@ -126,6 +126,11 @@ public void close(long timeoutMills) {
                 shutdownGraceful(timeoutMills);
             }
         }
+
+        // 注销MBean注册
+        if (this.forkJoinPoolMXBean != null) {
+            MBeanRegistryHelper.unregisterMBean(this.forkJoinPoolMXBean.getObjectName());
+        }
     }
 
     private void shutdownGraceful(long timeoutMills) {

trpc-core/src/main/java/com/tencent/trpc/core/worker/support/thread/ThreadWorkerPool.java

@@ -149,6 +149,10 @@ public void close(long timeoutMills) {
                 shutdownGraceful(timeoutMills);
             }
         }
+        // 注销MBean注册
+        if (this.threadPoolMXBean != null) {
+            MBeanRegistryHelper.unregisterMBean(this.threadPoolMXBean.getObjectName());
+        }
     }
 
     @Override