natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

【fix】forking cluster (#11063) 

* 【fix】forking cluster

* 【fix】fix code style

* 【fix】unit test unwrap exception

* 【fix】unit test unwrap CompletionException

* 【fix】code review

Co-authored-by: kanji.ainiwaer <kanji.ainiwaer@trip.com>
This commit is contained in:
Zeus 2022-12-14 15:40:21 +08:00 committed by GitHub
parent d17160591c
commit 116263e98d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 15 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/ForkingClusterInvoker.java
View File

@ -30,6 +30,8 @@ import org.apache.dubbo.rpc.cluster.LoadBalance;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@ -84,27 +86,28 @@ public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> {
RpcContext.getServiceContext().setInvokers((List) selected);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<>(1);
for (final Invoker<T> invoker : selected) {
selected.forEach(invoker -> {
URL consumerUrl = RpcContext.getServiceContext().getConsumerUrl();
executor.execute(() -> {
try {
if (ref.size() > 0) {
return;
}
Result result = invokeWithContextAsync(invoker, invocation, consumerUrl);
ref.offer(result);
} catch (Throwable e) {
CompletableFuture.<Object>supplyAsync(() -> {
if (ref.size() > 0) {
return null;
}
return invokeWithContextAsync(invoker, invocation, consumerUrl);
}, executor).whenComplete((v, t) -> {
if (t == null) {
ref.offer(v);
} else {
int value = count.incrementAndGet();
if (value >= selected.size()) {
ref.offer(e);
ref.offer(t);
}
}
});
}
});
try {
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
if (ret instanceof Throwable) {
Throwable e = (Throwable) ret;
Throwable e = ret instanceof CompletionException ? ((CompletionException) ret).getCause() : (Throwable) ret;
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : RpcException.UNKNOWN_EXCEPTION,
"Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. " +
"Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);