千家信息网

请输入关键字词

热门搜索排行

最新搜索排行

导航： 首页 > 互联网科技 >

dubbo中ForkingClusterInvoker的作用是什么

发表于：2025-12-02 作者：千家信息网编辑
千家信息网最后更新 2025年12月02日，这期内容当中小编将会给大家带来有关dubbo中ForkingClusterInvoker的作用是什么，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。ForkingCl
千家信息网最后更新 2025年12月02日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 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的作用是什么了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注行业资讯频道。


        


        
 很赞哦！ 

        


        



        作用
            个数
            内容
            参数
            结果
            选出
            分析
            专业
            两个
            中小
            内容丰富
            场景
            小结
            就是
            文章
            更多
            知识
            篇文章
            行业
            角度
    
数据库的安全要保护哪些东西
数据库安全各自的含义是什么
生产安全数据库录入
数据库的安全性及管理
数据库安全策略包含哪些
海淀数据库安全审计系统
建立农村房屋安全信息数据库
易用的数据库客户端支持安全管理
连接数据库失败ssl安全错误
数据库的锁怎样保障安全

杭州慧泽互联网科技有限公司
王者怎么寻找两三年前玩的服务器
福州软件开发薪水
陕汽的网络技术
战地5加入服务器怎么弄
网络安全月是几月
微星S1316服务器主板
数据库问题咨询
sql 获取整行数据库
武汉erp软件开发公司
天津慕诺网络技术有限公司欠薪
郧西公司软件开发包括哪些
国三网络技术成绩查询
时遇网络技术上海有限公司
网络安全考研推荐书
数学专业可以搞软件开发吗
任子行网络技术股份有
软件开发可以无形摊销吗
c 是软件开发哪个方面
服务器怎么识别宽带和数据
杭州慧泽互联网科技有限公司
中学生网络安全手抄报的句子
sql数据库的日期函数
微星S1316服务器主板
苏州想学软件开发怎么入手
杭州网络技术市场价
概念 数据库技术
华为手机登录不了邮箱服务器
北京仓储生鲜配送软件开发
艾斯威尔数据库推广
        

        


        
        

        

          
        

      
    


    



    

      

        
相关文章

      

      
    






   





  



































0