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

Fix RejectException (#12950) 

* fix RejectException null

* fix reference

* use isTerminated judge Executor

* use isShutdown judge Executor

---------

Co-authored-by: earthchen <earthchen1996@gmail.com>
This commit is contained in:
TomlongTK 2023-10-11 23:32:30 +08:00 committed by GitHub
parent 069fd5e1ce
commit c10d95b56a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 25 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/serial/SerializingExecutor.java
View File

@ -26,6 +26,7 @@ import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadlocal.InternalThreadLocalMap;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_ERROR_RUN_THREAD_TASK;
import static org.apache.dubbo.common.utils.ExecutorUtil.isShutdown;
/**
* Executor ensuring that all {@link Runnable} tasks submitted are executed in order
@ -68,8 +69,10 @@ public final class SerializingExecutor implements Executor, Runnable {
if (atomicBoolean.compareAndSet(false, true)) {
boolean success = false;
try {
executor.execute(this);
success = true;
if (!isShutdown(executor)) {
executor.execute(this);
success = true;
}
} finally {
// It is possible that at this point that there are still tasks in
// the queue, it would be nice to keep trying but the error may not

15
dubbo-common/src/main/java/org/apache/dubbo/common/utils/ExecutorUtil.java
View File

@ -38,12 +38,17 @@ public class ExecutorUtil {
new NamedThreadFactory("Close-ExecutorService-Timer", true));
public static boolean isTerminated(Executor executor) {
if (executor instanceof ExecutorService) {
if (((ExecutorService) executor).isTerminated()) {
return true;
}
if (!(executor instanceof ExecutorService)) {
return false;
}
return false;
return ((ExecutorService) executor).isTerminated();
}
public static boolean isShutdown(Executor executor) {
if (!(executor instanceof ExecutorService)) {
return false;
}
return ((ExecutorService) executor).isShutdown();
}
/**

1
dubbo-common/src/main/java/org/apache/dubbo/rpc/executor/AbstractIsolationExecutorSupport.java
View File

@ -36,6 +36,7 @@ public abstract class AbstractIsolationExecutorSupport implements ExecutorSuppor
this.frameworkServiceRepository = url.getOrDefaultFrameworkModel().getServiceRepository();
}
@Override
public Executor getExecutor(Object data) {
ProviderModel providerModel = getProviderModel(data);

1
dubbo-common/src/main/java/org/apache/dubbo/rpc/executor/DefaultExecutorSupport.java
View File

@ -30,6 +30,7 @@ public class DefaultExecutorSupport implements ExecutorSupport {
this.executorRepository = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel());
}
@Override
public Executor getExecutor(Object data) {
return executorRepository.getExecutor(url);
}

4
dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/AsyncRpcResult.java
View File

@ -180,7 +180,7 @@ public class AsyncRpcResult implements Result {
*/
@Override
public Result get() throws InterruptedException, ExecutionException {
if (executor != null && executor instanceof ThreadlessExecutor) {
if (executor instanceof ThreadlessExecutor) {
ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
try {
while (!responseFuture.isDone() && !threadlessExecutor.isShutdown()) {
@ -196,7 +196,7 @@ public class AsyncRpcResult implements Result {
@Override
public Result get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
long deadline = System.nanoTime() + unit.toNanos(timeout);
if (executor != null && executor instanceof ThreadlessExecutor) {
if (executor instanceof ThreadlessExecutor) {
ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
try {
while (!responseFuture.isDone() && !threadlessExecutor.isShutdown()) {

14
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DeadlineFuture.java
View File

@ -47,6 +47,9 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
private final List<Runnable> timeoutListeners = new ArrayList<>();
private final Timeout timeoutTask;
private ExecutorService executor;
private static final GlobalResourceInitializer<Timer> TIME_OUT_TIMER = new GlobalResourceInitializer<>(
() -> new HashedWheelTimer(new NamedThreadFactory("dubbo-future-timeout", true), 30,
TimeUnit.MILLISECONDS), DeadlineFuture::destroy);
private DeadlineFuture(String serviceName, String methodName, String address, int timeout) {
this.serviceName = serviceName;
@ -76,20 +79,15 @@ public class DeadlineFuture extends CompletableFuture<AppResponse> {
}
public void received(TriRpcStatus status, AppResponse appResponse) {
if (status.code != TriRpcStatus.Code.DEADLINE_EXCEEDED) {
// decrease Time
if (!timeoutTask.isCancelled()) {
timeoutTask.cancel();
}
if (status.code != TriRpcStatus.Code.DEADLINE_EXCEEDED && !timeoutTask.isCancelled()) {
timeoutTask.cancel();
}
if (getExecutor() != null) {
getExecutor().execute(() -> doReceived(status, appResponse));
} else {
doReceived(status, appResponse);
}
} private static final GlobalResourceInitializer<Timer> TIME_OUT_TIMER = new GlobalResourceInitializer<>(
() -> new HashedWheelTimer(new NamedThreadFactory("dubbo-future-timeout", true), 30,
TimeUnit.MILLISECONDS), DeadlineFuture::destroy);
}
public void addTimeoutListener(Runnable runnable) {
timeoutListeners.add(runnable);