序

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