聊聊dubbo的ForkingClusterInvoker

序

本文主要研究一下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

doc

• ForkingClusterInvoker