java - 工作线程的线程池任务执行器超时

Question

我正在使用 spring boot 并且有一种异步方法。要执行异步我有以下配置，问题是如果所有这 5 个线程由于某种原因挂起怎么办，基本上它会锁定应用程序并且不会执行任何新任务(它只会继续接受)。我们如何为那些工作线程设置超时，比如说 120 秒，然后超时并执行新任务。 (是的，我正在使用带有无界队列的固定线程池来继续接受任务)

@EnableAsync
@Configuration
public class AsyncConfiguration implements AsyncConfigurer {

@Override
public Executor getAsyncExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setCorePoolSize(5);
    taskExecutor.setMaxPoolSize(5);
    taskExecutor.initialize();
    return taskExecutor;
}

@Override
public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
    return new SimpleAsyncUncaughtExceptionHandler();
}

Answer 1

您可以创建另一个执行器，例如:

static class TimeOutExecutorService extends CompletableExecutors.DelegatingCompletableExecutorService {
    private final Duration timeout;
    private final ScheduledExecutorService schedulerExecutor;

    TimeOutExecutorService(ExecutorService delegate, Duration timeout) {
        super(delegate);
        this.timeout = timeout;
        schedulerExecutor = Executors.newScheduledThreadPool(1);
    }

    @Override public <T> CompletableFuture<T> submit(Callable<T> task) {
        CompletableFuture<T> cf = new CompletableFuture<>();
        Future<?> future = delegate.submit(() -> {
            try {
                cf.complete(task.call());
            } catch (CancellationException e) {
                cf.cancel(true);
            } catch (Throwable ex) {
                cf.completeExceptionally(ex);
            }
        });

        schedulerExecutor.schedule(() -> {
            if (!cf.isDone()) {
                cf.completeExceptionally(new TimeoutException("Timeout after " + timeout));
                future.cancel(true);
            }
        }, timeout.toMillis(), TimeUnit.MILLISECONDS);
        return cf;
    }
}

然后，创建一个名为 timed 的新 bean

@Bean(name = "timed")
public Executor timeoutExecutor() {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("timed-%d").build();
    return TimedCompletables.timed(Executors.newFixedThreadPool(10, threadFactory), Duration.ofSeconds(2));
}

然后，尝试使用此 Executor 来执行您的异步任务。

或者，尝试将您的代码从 FixSizeThreadPool 更改为构建自己的线程池执行器。