получить вызов от ThreadPoolTaskExecutor или сделать Runnable to Callable

Question

Я использую ThreadPoolTaskExecutor для выполнения своих задач, которые являются реализациями интерфейса Callable. Я просто хочу проверить время, если задача все еще находится в пуле (мониторинг). Как это сделать? Я знаю, что я могу получить очередь из ThreadPoolExecutor, но как я могу использовать Runnable to Callable?

В принципе я это вызываемая

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

Я его выполнения, как этот

ThreadPoolExecutor.submit(new Formatter(order)); 

И, наконец, я хотел бы Переберите очереди ExecutorService в какой-то метод асинхронной и если нить проверить с orderId все еще существует.

Answer 1

Как объяснено в this answer, вы можете получить контроль над FutureTask обертывание Callable путем создания его вручную и enqueuing через 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(); 
    } 
} 

enqueuing его через threadPoolExecutor.execute(new MyFutureTask(new Formatter(order)));,

вы можете запросить идентификаторы заказов на очереди:

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, а общий метод генерирует непроверенное предупреждение для специального лечения.

Answer 2

Поскольку, похоже, вы хотите контролировать ExecutorService, посмотрите на переопределение decorateTask(). Затем вы можете украсить будущее, чтобы контролировать его состояние.