Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Use InternalThreadLocal in consumer side #1825

Merged
merged 19 commits into from
May 25, 2018
Merged

Use InternalThreadLocal in consumer side #1825

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
0807761
SerializerFactory 获取Serializer时，锁住整个hashmap，导致整个过程被block
carryxyh Jan 10, 2018
65438b4
单元测试。保证一个class只有一个serializer和deserializer。单线程和多线程测试
carryxyh Jan 10, 2018
7291963
增加线程数 50 模拟多个线程来获取serializer和deserializer
carryxyh Jan 10, 2018
ed2ff0f
当cores线程数全都使用的情况下，默认线程池会把任务放入到队列中。队列满则再创建线程（总数不会超过Max线程数）
carryxyh Jan 11, 2018
f97db80
当cores线程数全都使用的情况下，默认线程池会把任务放入到队列中。队列满则再创建线程（总数不会超过Max线程数）
carryxyh Jan 11, 2018
a224bb0
Merge remote-tracking branch 'upstream/master'
carryxyh Jan 12, 2018
49aecfb
补全单元测试，测试扩展是否生效
carryxyh Jan 17, 2018
561397f
Merge remote-tracking branch 'upstream/master'
carryxyh Feb 7, 2018
60c2993
错误命名
carryxyh Feb 7, 2018
af29757
Merge remote-tracking branch 'upstream/master'
carryxyh Feb 26, 2018
4be8d61
增加@Override注解
carryxyh Feb 27, 2018
5cfe610
修复单元测试
carryxyh Feb 27, 2018
7830079
Merge remote-tracking branch 'upstream/master'
carryxyh Apr 8, 2018
a1a42c0
Merge remote-tracking branch 'upstream/master'
carryxyh Apr 25, 2018
be1418f
remove enhanced
carryxyh Apr 25, 2018
9653553
remove enhanced
carryxyh Apr 25, 2018
e28aa17
Merge remote-tracking branch 'upstream/master'
carryxyh May 5, 2018
3896eaf
Merge remote-tracking branch 'upstream/master'
carryxyh May 21, 2018
9532e1f
Change ThreadFactory for consumer side which is to use InternalThread…
carryxyh May 21, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions ...o-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailbackClusterInvoker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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;

Expand Down
9 changes: 7 additions & 2 deletions dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/ForkingClusterInvoker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
Expand Down