这期内容当中小编将会给大家带来有关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();
}
}
} 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);
}
} 小结 ForkingClusterInvoker使用Executors.newCachedThreadPool创建了一个executor；其doInvoke从url获取forks及timeout参数，然后从invokers选出forks个数的invoker，然后放到executor请求执行invoker.invoke(invocation)，把Result放到LinkedBlockingQueue，最后使用指定的timeout去poll出第一个返回结果返回，异常的话抛出RpcException
上述就是小编为大家分享的dubbo中ForkingClusterInvoker的作用是什么了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注创新互联行业资讯频道。
分享文章：dubbo中ForkingClusterInvoker的作用是什么
URL地址：
http://cdweb.net/article/pjjdpo.html
