序
本文主要研究一下dubbo的ForkingClusterInvoker
ForkingClusterInvoker
dubbo-2.7.3/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java
public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
/**
* Use {@link NamedInternalThreadFactory} to produce {@link org.apache.dubbo.common.threadlocal.InternalThread}
* which with the use of {@link org.apache.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);
}
@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
final List<Invoker<T>> selected;
final int forks = getUrl().getParameter(FORKS_KEY, DEFAULT_FORKS);
final int timeout = getUrl().getParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
selected = new ArrayList<>();
for (int i = 0; i < forks; i++) {
Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
if (!selected.contains(invoker)) {
//Avoid add the same invoker several times.
selected.add(invoker);
}
}
}
RpcContext.getContext().setInvokers((List) selected);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>();
for (final Invoker<T> invoker : selected) {
executor.execute(new Runnable() {
@Override
public void run() {
try {
Result result = invoker.invoke(invocation);
ref.offer(result);
} catch (Throwable e) {
int value = count.incrementAndGet();
if (value >= selected.size()) {
ref.offer(e);
}
}
}
});
}
try {
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
if (ret instanceof Throwable) {
Throwable e = (Throwable) ret;
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
}
return (Result) ret;
} catch (InterruptedException e) {
throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
}
} finally {
// clear attachments which is binding to current thread.
RpcContext.getContext().clearAttachments();
}
}
}- ForkingClusterInvoker使用Executors.newCachedThreadPool创建了一个executor；其doInvoke从url获取forks及timeout参数，然后从invokers选出forks个数的invoker，然后放到executor请求执行invoker.invoke(invocation)，把Result放到LinkedBlockingQueue，最后使用指定的timeout去poll出第一个返回结果返回，异常的话抛出RpcException
ForkingClusterInvokerTest
dubbo-2.7.3/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvokerTest.java
public class ForkingClusterInvokerTest {
private List<Invoker<ForkingClusterInvokerTest>> invokers = new ArrayList<Invoker<ForkingClusterInvokerTest>>();
private URL url = URL.valueOf("test://test:11/test?forks=2");
private Invoker<ForkingClusterInvokerTest> invoker1 = mock(Invoker.class);
private Invoker<ForkingClusterInvokerTest> invoker2 = mock(Invoker.class);
private Invoker<ForkingClusterInvokerTest> invoker3 = mock(Invoker.class);
private RpcInvocation invocation = new RpcInvocation();
private Directory<ForkingClusterInvokerTest> dic;
private Result result = new AppResponse();
@BeforeEach
public void setUp() throws Exception {
dic = mock(Directory.class);
given(dic.getUrl()).willReturn(url);
given(dic.list(invocation)).willReturn(invokers);
given(dic.getInterface()).willReturn(ForkingClusterInvokerTest.class);
invocation.setMethodName("method1");
invokers.add(invoker1);
invokers.add(invoker2);
invokers.add(invoker3);
}
private void resetInvokerToException() {
given(invoker1.invoke(invocation)).willThrow(new RuntimeException());
given(invoker1.getUrl()).willReturn(url);
given(invoker1.isAvailable()).willReturn(true);
given(invoker1.getInterface()).willReturn(ForkingClusterInvokerTest.class);
given(invoker2.invoke(invocation)).willThrow(new RuntimeException());
given(invoker2.getUrl()).willReturn(url);
given(invoker2.isAvailable()).willReturn(true);
given(invoker2.getInterface()).willReturn(ForkingClusterInvokerTest.class);
given(invoker3.invoke(invocation)).willThrow(new RuntimeException());
given(invoker3.getUrl()).willReturn(url);
given(invoker3.isAvailable()).willReturn(true);
given(invoker3.getInterface()).willReturn(ForkingClusterInvokerTest.class);
}
private void resetInvokerToNoException() {
given(invoker1.invoke(invocation)).willReturn(result);
given(invoker1.getUrl()).willReturn(url);
given(invoker1.isAvailable()).willReturn(true);
given(invoker1.getInterface()).willReturn(ForkingClusterInvokerTest.class);
given(invoker2.invoke(invocation)).willReturn(result);
given(invoker2.getUrl()).willReturn(url);
given(invoker2.isAvailable()).willReturn(true);
given(invoker2.getInterface()).willReturn(ForkingClusterInvokerTest.class);
given(invoker3.invoke(invocation)).willReturn(result);
given(invoker3.getUrl()).willReturn(url);
given(invoker3.isAvailable()).willReturn(true);
given(invoker3.getInterface()).willReturn(ForkingClusterInvokerTest.class);
}
@Test
public void testInvokeException() {
resetInvokerToException();
ForkingClusterInvoker<ForkingClusterInvokerTest> invoker = new ForkingClusterInvoker<ForkingClusterInvokerTest>(
dic);
try {
invoker.invoke(invocation);
Assertions.fail();
} catch (RpcException expected) {
Assertions.assertTrue(expected.getMessage().contains("Failed to forking invoke provider"));
assertFalse(expected.getCause() instanceof RpcException);
}
}
@Test
public void testClearRpcContext() {
resetInvokerToException();
ForkingClusterInvoker<ForkingClusterInvokerTest> invoker = new ForkingClusterInvoker<ForkingClusterInvokerTest>(
dic);
String attachKey = "attach";
String attachValue = "value";
RpcContext.getContext().setAttachment(attachKey, attachValue);
Map<String, String> attachments = RpcContext.getContext().getAttachments();
Assertions.assertTrue(attachments != null && attachments.size() == 1, "set attachment failed!");
try {
invoker.invoke(invocation);
Assertions.fail();
} catch (RpcException expected) {
Assertions.assertTrue(expected.getMessage().contains("Failed to forking invoke provider"), "Succeeded to forking invoke provider !");
assertFalse(expected.getCause() instanceof RpcException);
}
Map<String, String> afterInvoke = RpcContext.getContext().getAttachments();
Assertions.assertTrue(afterInvoke != null && afterInvoke.size() == 0, "clear attachment failed!");
}
@Test()
public void testInvokeNoException() {
resetInvokerToNoException();
ForkingClusterInvoker<ForkingClusterInvokerTest> invoker = new ForkingClusterInvoker<ForkingClusterInvokerTest>(
dic);
Result ret = invoker.invoke(invocation);
Assertions.assertSame(result, ret);
}
}- ForkingClusterInvokerTest验证了testInvokeException、testClearRpcContext两个场景
小结
ForkingClusterInvoker使用Executors.newCachedThreadPool创建了一个executor；其doInvoke从url获取forks及timeout参数，然后从invokers选出forks个数的invoker，然后放到executor请求执行invoker.invoke(invocation)，把Result放到LinkedBlockingQueue，最后使用指定的timeout去poll出第一个返回结果返回，异常的话抛出RpcException
