dubbo中ForkingClusterInvoker的作用是什么

這期內容當中小編將會給大家?guī)碛嘘Pdubbo中ForkingClusterInvoker的作用是什么，文章內容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

創(chuàng)新互聯(lián)專業(yè)為企業(yè)提供烏什網(wǎng)站建設、烏什做網(wǎng)站、烏什網(wǎng)站設計、烏什網(wǎng)站制作等企業(yè)網(wǎng)站建設、網(wǎng)頁設計與制作、烏什企業(yè)網(wǎng)站模板建站服務，10年烏什做網(wǎng)站經驗，不只是建網(wǎng)站，更提供有價值的思路和整體網(wǎng)絡服務。

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創(chuàng)建了一個executor；其doInvoke從url獲取forks及timeout參數(shù)，然后從invokers選出forks個數(shù)的invoker，然后放到executor請求執(zhí)行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創(chuàng)建了一個executor；其doInvoke從url獲取forks及timeout參數(shù)，然后從invokers選出forks個數(shù)的invoker，然后放到executor請求執(zhí)行invoker.invoke(invocation)，把Result放到LinkedBlockingQueue，最后使用指定的timeout去poll出第一個返回結果返回，異常的話拋出RpcException

上述就是小編為大家分享的dubbo中ForkingClusterInvoker的作用是什么了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注創(chuàng)新互聯(lián)行業(yè)資訊頻道。

當前標題：dubbo中ForkingClusterInvoker的作用是什么
網(wǎng)站鏈接：http://muchs.cn/article30/pjjdpo.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供云服務器、服務器托管、微信小程序、用戶體驗、網(wǎng)站策劃、網(wǎng)站導航

廣告

聲明：本網(wǎng)站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)

成都做網(wǎng)站