Use InternalThreadLocal in consumer side (#1825)

* SerializerFactory 获取Serializer时，锁住整个hashmap，导致整个过程被block * 单元测试。保证一个class只有一个serializer和deserializer。单线程和多线程测试 * 增加线程数 50 模拟多个线程来获取serializer和deserializer * 当cores线程数全都使用的情况下，默认线程池会把任务放入到队列中。队列满则再创建线程（总数不会超过Max线程数）增强线程池：在请求量阶段性出现高峰时使用特性：cores线程全部使用的情况下，优先创建线程（总数不会超过max），当max个线程全都在忙的情况下，才将任务放入队列。请求量下降时，线程池会自动维持cores个线程，多余的线程退出。 * 当cores线程数全都使用的情况下，默认线程池会把任务放入到队列中。队列满则再创建线程（总数不会超过Max线程数）增强线程池：在请求量阶段性出现高峰时使用特性：cores线程全部使用的情况下，优先创建线程（总数不会超过max），当max个线程全都在忙的情况下，才将任务放入队列。请求量下降时，线程池会自动维持cores个线程，多余的线程退出。 * 补全单元测试，测试扩展是否生效 * 错误命名 * 增加@OverRide注解 long 初始化赋值时，小写l改为大写L防止误读 * 修复单元测试 * remove enhanced * remove enhanced * Change ThreadFactory for consumer side which is to use InternalThreadLocal in RpcContext.
apache · May 25, 2018 · 0bf6910 · 0bf6910
1 parent 47adfd0
commit 0bf6910
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 4 deletions.
diff --git a/...o-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java b/...o-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java
@@ -18,12 +18,13 @@
 
 import com.alibaba.dubbo.common.logger.Logger;
 import com.alibaba.dubbo.common.logger.LoggerFactory;
-import com.alibaba.dubbo.common.utils.NamedThreadFactory;
+import com.alibaba.dubbo.common.threadlocal.NamedInternalThreadFactory;
 import com.alibaba.dubbo.rpc.Invocation;
 import com.alibaba.dubbo.rpc.Invoker;
 import com.alibaba.dubbo.rpc.Result;
 import com.alibaba.dubbo.rpc.RpcException;
 import com.alibaba.dubbo.rpc.RpcResult;
+import com.alibaba.dubbo.rpc.RpcContext;
 import com.alibaba.dubbo.rpc.cluster.Directory;
 import com.alibaba.dubbo.rpc.cluster.LoadBalance;
 
@@ -50,7 +51,13 @@ public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
     private static final long RETRY_FAILED_PERIOD = 5 * 1000;
 
-    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2, new NamedThreadFactory("failback-cluster-timer", true));
+    /**
+     * Use {@link NamedInternalThreadFactory} to produce {@link com.alibaba.dubbo.common.threadlocal.InternalThread}
+     * which with the use of {@link com.alibaba.dubbo.common.threadlocal.InternalThreadLocal} in {@link RpcContext}.
+     */
+    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2,
+            new NamedInternalThreadFactory("failback-cluster-timer", true));
+
     private final ConcurrentMap<Invocation, AbstractClusterInvoker<?>> failed = new ConcurrentHashMap<Invocation, AbstractClusterInvoker<?>>();
     private volatile ScheduledFuture<?> retryFuture;
 

diff --git a/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ForkingClusterInvoker.java b/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ForkingClusterInvoker.java
@@ -17,7 +17,7 @@
 package com.alibaba.dubbo.rpc.cluster.support;
 
 import com.alibaba.dubbo.common.Constants;
-import com.alibaba.dubbo.common.utils.NamedThreadFactory;
+import com.alibaba.dubbo.common.threadlocal.NamedInternalThreadFactory;
 import com.alibaba.dubbo.rpc.Invocation;
 import com.alibaba.dubbo.rpc.Invoker;
 import com.alibaba.dubbo.rpc.Result;
@@ -43,7 +43,12 @@
  */
 public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
-    private final ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("forking-cluster-timer", true));
+    /**
+     * Use {@link NamedInternalThreadFactory} to produce {@link com.alibaba.dubbo.common.threadlocal.InternalThread}
+     * which with the use of {@link com.alibaba.dubbo.common.threadlocal.InternalThreadLocal} in {@link RpcContext}.
+     */
+    private final ExecutorService executor = Executors.newCachedThreadPool(
+            new NamedInternalThreadFactory("forking-cluster-timer", true));
 
     public ForkingClusterInvoker(Directory<T> directory) {
         super(directory);