从 ThreadPoolTask​​Executor 获取可调用对象或将 Runnable 转换为 Callable

我在用着线程池任务执行器用于执行我的任务，这些任务是Callable界面。我只想及时检查任务是否仍在池中（监视）。怎么做？我知道我可以从线程池执行器但我怎样才能将 Runnable 转换为 Callable 呢？

基本上我有这个可调用的

public interface IFormatter extends Callable<Integer>{
    Long getOrderId();
}

我像这样执行

ThreadPoolExecutor.submit(new Formatter(order));

最后，我想在一些异步方法中循环遍历 ExecutorService 队列，并检查具有 orderId 的线程是否仍然存在。

正如中所解释的这个答案，你可以控制FutureTask包裹着Callable通过手动创建它并通过排队execute。否则，submit将包裹你的Callable进入一个ExecutorService- 特定对象并将其放入队列中，从而无法查询该对象的属性Callable通过标准 API。

使用自定义FutureTask

class MyFutureTask extends FutureTask<Integer> {
    final IFormatter theCallable;

    public MyFutureTask(IFormatter callable) {
        super(callable);
        theCallable=callable;
    }
    Long getOrderId() {
        return theCallable.getOrderId();
    }
}

通过将其排队threadPoolExecutor.execute(new MyFutureTask(new Formatter(order)));,

您可以查询队列中的订单ID：

public static boolean isEnqueued(ThreadPoolExecutor e, Long id) {
    for(Object o: e.getQueue().toArray()) {
        if(o instanceof MyFutureTask && Objects.equals(((MyFutureTask)o).getOrderId(), id))
            return true;
    }
    return false;
}

这适用于任何ExecutorService（假设它有一个队列）。如果您使用的是ThreadPoolExecutor只是，您可以自定义其创建FutureTask实例（从 Java 6 开始），而不是依赖提交者执行此操作：

public class MyThreadPoolExecutor extends ThreadPoolExecutor {

    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                                TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
        TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
            workQueue, threadFactory);
    }
    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
        TimeUnit unit, BlockingQueue<Runnable> workQueue,
        RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
            workQueue, handler);
    }
    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
        TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
        RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit,
            workQueue, threadFactory, handler);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if(callable instanceof IFormatter)
            return (FutureTask<T>)new MyFutureTask((IFormatter)callable);
        return super.newTaskFor(callable);
    }
}

然后，使用一个实例MyThreadPoolExecutor代替ThreadPoolExecutor每提交一个IFormatter实例将自动包装使用MyFutureTask而不是标准FutureTask。缺点是这只适用于这个特定的ExecutorService通用方法会针对特殊处理生成未经检查的警告。