是否存在Java Future实现可在线程池繁忙时于调用线程执行任务？

针对你遇到的线程池被阻塞请求占满的问题，我们可以通过自定义ExecutorService的方式实现“未启动任务在Future.get()时由调用线程执行”的逻辑，作为服务的安全阀，避免整体卡顿。以下是具体方案：

自定义ExecutorService包装类 #

这是最直接的实现方式，通过包装现有ExecutorService，自定义FutureTask来跟踪任务状态，在调用get()时触发调用线程执行未启动的任务。

核心实现代码 #

import java.util.concurrent.*;

// 自定义FutureTask，跟踪任务是否已启动
class CallerRunsOnGetFutureTask<V> extends FutureTask<V> {
    private volatile boolean started;

    public CallerRunsOnGetFutureTask(Callable<V> callable) {
        super(callable);
    }

    @Override
    protected void run() {
        started = true;
        super.run();
    }

    @Override
    public V get() throws InterruptedException, ExecutionException {
        // 任务未启动且未完成时，调用线程直接执行
        if (!started && !isDone()) {
            run();
        }
        return super.get();
    }

    @Override
    public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        if (!started && !isDone()) {
            run();
        }
        return super.get(timeout, unit);
    }
}

// 包装ExecutorService，提交任务时使用自定义FutureTask
class CallerRunsOnGetExecutor implements ExecutorService {
    private final ExecutorService delegate;

    public CallerRunsOnGetExecutor(ExecutorService delegate) {
        this.delegate = delegate;
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        CallerRunsOnGetFutureTask<T> futureTask = new CallerRunsOnGetFutureTask<>(task);
        delegate.execute(futureTask);
        return futureTask;
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return submit(Executors.callable(task, result));
    }

    @Override
    public Future<?> submit(Runnable task) {
        return submit(Executors.callable(task));
    }

    // 其余ExecutorService方法直接委托给底层线程池实现
    @Override public void shutdown() { delegate.shutdown(); }
    @Override public List<Runnable> shutdownNow() { return delegate.shutdownNow(); }
    @Override public boolean isShutdown() { return delegate.isShutdown(); }
    @Override public boolean isTerminated() { return delegate.isTerminated(); }
    @Override public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return delegate.awaitTermination(timeout, unit);
    }
    @Override public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        return delegate.invokeAll(tasks);
    }
    @Override public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        return delegate.invokeAll(tasks, timeout, unit);
    }
    @Override public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        return delegate.invokeAny(tasks);
    }
    @Override public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return delegate.invokeAny(tasks, timeout, unit);
    }
    @Override public void execute(Runnable command) { delegate.execute(command); }
}

使用方式 #

创建普通线程池后，用自定义包装类包裹即可：

// 初始化底层线程池，根据业务设置核心/最大线程数、队列大小
ThreadPoolExecutor basePool = new ThreadPoolExecutor(
        4, 8,
        60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(32)
);
ExecutorService callerRunsExecutor = new CallerRunsOnGetExecutor(basePool);

// 提交云存储查询任务
Future<Boolean> blobExists = callerRunsExecutor.submit(() -> checkBlobExists(path));
Future<Boolean> dirBlobExists = callerRunsExecutor.submit(() -> checkDirBlobExists(path));
Future<Boolean> hasPrefixBlobs = callerRunsExecutor.submit(() -> checkPrefixBlobs(path));

// 调用get()时，若任务未启动则由当前线程执行
boolean exists = blobExists.get();

关键注意事项 #

1. 线程安全：用volatile修饰started变量，确保多线程下的状态可见性
2. 异常兼容：任务执行时抛出的异常会被Future正常捕获，和线程池执行逻辑一致
3. 性能平衡：当线程池繁忙时，调用线程会被占用执行任务，单个请求响应时间会小幅增加，但避免了整个服务因线程池耗尽而全面卡顿，完全匹配你的云存储查询场景需求
4. 复用现有线程池：底层可以用任意标准ExecutorService实现，无需完全重写线程池逻辑

内容的提问来源于stack exchange，提问作者Wheezil