You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by il...@apache.org on 2018/05/25 02:14:06 UTC
[incubator-dubbo] branch master updated: Use InternalThreadLocal in
consumer side (#1825)
This is an automated email from the ASF dual-hosted git repository.
iluo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-dubbo.git
The following commit(s) were added to refs/heads/master by this push:
new 0bf6910 Use InternalThreadLocal in consumer side (#1825)
0bf6910 is described below
commit 0bf6910b672add6363004010ab6421158c7097dd
Author: 时无两丶 <44...@qq.com>
AuthorDate: Fri May 25 10:14:03 2018 +0800
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.
---
.../dubbo/rpc/cluster/support/FailbackClusterInvoker.java | 11 +++++++++--
.../dubbo/rpc/cluster/support/ForkingClusterInvoker.java | 9 +++++++--
2 files changed, 16 insertions(+), 4 deletions(-)
diff --git a/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java b/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java
index 17b50ff..3cc22a8 100644
--- a/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java
+++ b/dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java
@@ -18,12 +18,13 @@ package com.alibaba.dubbo.rpc.cluster.support;
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
index dd6b901..7d9a58d 100644
--- 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 @@ import java.util.concurrent.atomic.AtomicInteger;
*/
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);
--
To stop receiving notification emails like this one, please contact
iluo@apache.org.