咨询热线 ：
028-86922220
×
01
关于我们

公司简介

发展历程

02
服务项目

高端网站建设

微信开发

APP开发

网络营销服务

电商网站定制

生物医药网站建设

外贸网站建设

教育培训网站建设

03
小程序开发

小程序开发

功能应用

客户案例

04
案例展示

网站建设案例

小程序案例

电商平台案例

APP案例

系统平台案例

05
建站动态

网站建设

网站设计

常见问题

小程序

06
联系我们

公司地址

人才招聘

地址：成都市太升南路288号锦天国际A幢1002号

电话：028-86922220

028-86922220

重庆分公司，新征程启航

为企业提供网站建设、域名注册、服务器等服务

dubbo中ForkingClusterInvoker的作用是什么

这期内容当中小编将会给大家带来有关dubbo中ForkingClusterInvoker的作用是什么，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。

创新互联专业为企业提供乌什网站建设、乌什做网站、乌什网站设计、乌什网站制作等企业网站建设、网页设计与制作、乌什企业网站模板建站服务，10年乌什做网站经验，不只是建网站，更提供有价值的思路和整体网络服务。

ForkingClusterInvoker

dubbo-2.7.3/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java

public class ForkingClusterInvoker extends AbstractClusterInvoker {

    /**
     * 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 directory) {
        super(directory);
    }

    @Override
    @SuppressWarnings({"unchecked", "rawtypes"})
    public Result doInvoke(final Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {
        try {
            checkInvokers(invokers, invocation);
            final List> 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 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 ref = new LinkedBlockingQueue<>();
            for (final Invoker 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> invokers = new ArrayList>();
    private URL url = URL.valueOf("test://test:11/test?forks=2");
    private Invoker invoker1 = mock(Invoker.class);
    private Invoker invoker2 = mock(Invoker.class);
    private Invoker invoker3 = mock(Invoker.class);
    private RpcInvocation invocation = new RpcInvocation();
    private Directory 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 invoker = new ForkingClusterInvoker(
                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 invoker = new ForkingClusterInvoker(
                dic);

        String attachKey = "attach";
        String attachValue = "value";

        RpcContext.getContext().setAttachment(attachKey, attachValue);

        Map 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 afterInvoke = RpcContext.getContext().getAttachments();
        Assertions.assertTrue(afterInvoke != null && afterInvoke.size() == 0, "clear attachment failed!");
    }

    @Test()
    public void testInvokeNoException() {

        resetInvokerToNoException();

        ForkingClusterInvoker invoker = new ForkingClusterInvoker(
                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
上述就是小编为大家分享的dubbo中ForkingClusterInvoker的作用是什么了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注创新互联行业资讯频道。
            
            
                        

            文章标题：dubbo中ForkingClusterInvoker的作用是什么            

            当前网址：http://cqcxhl.cn/article/pjjdpo.html
        
    
    

        
其他资讯

        
    




    

        

            

                

                

                    
分享至：

                    

                        
                        
                        
                        
                    

                    
                

            

            

                

                    
关于我们

                    

                        公司简介
                        发展历程
                    

                

                

                    
服务项目

                    

                        高端网站建设
                        小程序开发
                        APP开发
                        网络营销
                    

                

                

                    
建站知识

                    

                        网站建设
                        网站设计
                        网站制作
                        小程序
                    

                

                

                    
联系我们

                    

                        公司地址
                        人才招聘
                    

                

            

            

                

                    成都
                    重庆
                

                

                    

                        

                            

                            
电话：028-86922220

                        

                        

                            

                            
地址：成都市太升南路288号锦天国际A幢1002号

                        

                    

                    

                        

                            

                            
电话：028-86922220

                        

                        

                            

                            
地址：重庆市南岸区弹子石腾龙大道58号2栋21-6

                        

                    

                

            

        

    

    
版权所有：成都创新互联科技有限公司重庆分公司
        渝ICP备2021005571号-4
    







    
        

            

            
在线咨询

        

    
    
        

            

            
服务热线

        

        
服务热线：028-86922220

    
    
        

            

            
TOP