从ThreadPoolTask​​Executor获取可调用性或将Runnable强制转换为Callable [英] get callable from ThreadPoolTaskExecutor or cast Runnable to Callable

问题描述

我正在使用 ThreadPoolTask​​Executor 用于执行我的任务，这些任务是 Callable 界面。我只想检查任务是否仍在池中（监控）。怎么做？我知道我可以从 ThreadPoolExecutor 但是如何将Runnable转换为Callable？

I'm using ThreadPoolTaskExecutor for executing my tasks which are implemantations of Callable interface. I just want to check in time if task is still in pool (monitoring). How to do that? I know that I can get queue from ThreadPoolExecutor but how can I cast Runnable to Callable?

基本上我有这个可调用的

Basically I have this callable

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

我正在执行它

ThreadPoolExecutor.submit(new Formatter(order));

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

And finally I'd like to loop through queue of ExecutorService in some async method and check if thread with orderId is still there.

推荐答案

如这个答案，您可以通过手动创建它来控制 FutureTask 包装 Callable 并通过执行进行排队。否则， submit 会将 Callable 包装到 ExecutorService 中 - 特定对象并将其放入队列，因此无法通过标准API查询 Callable 的属性。

As explained in this answer, you may get control over the FutureTask wrapping the Callable by creating it manually and enqueuing via execute. Otherwise, submit will wrap your Callable into an ExecutorService-specific object and put it into the queue, making it impossible to query properties of the Callable via standard APIs.

使用自定义 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：

you can query order IDs on the queue:

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开始），而不是依赖提交者这样做：

This works for any ExecutorService (assuming it has a queue). If you are using a ThreadPoolExecutor only, you may customize its creation of FutureTask instance (starting with Java 6), instead of relying on the submitter doing it:

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 实例的每次提交都将使用<$自动换行c $ c> MyFutureTask 而不是标准 FutureTask 。缺点是这只适用于这个特定的 ExecutorService ，并且泛型方法会为特殊处理生成未经检查的警告。

Then, using an instance of MyThreadPoolExecutor instead of ThreadPoolExecutor every submission of an IFormatter instance will automatically wrapped using MyFutureTask instead of the standard FutureTask. The drawback is that this works only with this specific ExecutorService and the generic method generates an unchecked warning for the special treatment.

这篇关于从ThreadPoolTask​​Executor获取可调用性或将Runnable强制转换为Callable的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！