这期内容当中小编将会给大家带来有关dubbo中ForkingClusterInvoker的作用是什么,文章内容丰富且以专业的角度为大家分析和叙述,阅读完这篇文章希望大家可以有所收获。
创新互联专业为企业提供乌什网站建设、乌什做网站、乌什网站设计、乌什网站制作等企业网站建设、网页设计与制作、乌什企业网站模板建站服务,10年乌什做网站经验,不只是建网站,更提供有价值的思路和整体网络服务。
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
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
上述就是小编为大家分享的dubbo中ForkingClusterInvoker的作用是什么了,如果刚好有类似的疑惑,不妨参照上述分析进行理解。如果想知道更多相关知识,欢迎关注创新互联行业资讯频道。
当前标题:dubbo中ForkingClusterInvoker的作用是什么
网站链接:https://www.cdcxhl.com/article30/pjjdpo.html
成都网站建设公司_创新互联,为您提供云服务器、服务器托管、微信小程序、用户体验、网站策划、网站导航
声明:本网站发布的内容(图片、视频和文字)以用户投稿、用户转载内容为主,如果涉及侵权请尽快告知,我们将会在第一时间删除。文章观点不代表本网站立场,如需处理请联系客服。电话:028-86922220;邮箱:631063699@qq.com。内容未经允许不得转载,或转载时需注明来源: 创新互联