Come posso utilizzare PriorityBlockingQueue con ListeningExecutorService?

Question

Poiché Guava's ListeningExecutorService è implementato avvolgendo un ExecutorService esistente, "decora" l'attività intercettando il metodo execute(). Ciò significa che se voglio utilizzare un PriorityQueue personalizzato sul ExecutorService sottostante, il mio comparatore "vede" l'attività decorata come oggetto ListenableFutureTask invece dell'originale.

C'è un modo per afferrare l'attività che esegue? In modo che il comparatore della coda possa utilizzare il peso delle attività per determinare l'ordine?

Answer 1

presumo che sei preoccupato con submit() piuttosto che execute()? (Vedere la parte inferiore della mia risposta.)

Con un ListeningExecutorService da MoreExecutors.listeningDecorator (il tipo di wrapper a cui si fa riferimento), sei sfortunato. listeningDecorator, come la maggior parte delle implementazioni ExecutorService, inserisce qualsiasi input su submit in un FutureTask. La soluzione normale a questo problema è implementare AbstractExecutorService e sostituire newTaskFor per restituire un oggetto personalizzato. Questo dovrebbe funzionare anche qui. Fondamentalmente si sta reimplementando listeningDecorator, che è un wrapper abbastanza banale intorno a AbstractListeningExecutorService, che è esso stesso un involucro abbastanza banale intorno a AbstractExecutorService.

Ci sono due complicazioni di coppia. (OK, ci potrebbe essere di più. Ammetto che non ho ancora testato l'approccio che sto suggerendo.)

1. AbstractListeningExecutorService non consente di ignorare newTaskFor. (Perché? Posso spiegare se desideri file a feature request.) Di conseguenza, devi estendere AbstractExecutorService direttamente, in gran parte duplicando l'implementazione (breve) AbstractListeningExecutorService.
2. newTaskFor deve restituire un ListenableFuture che è anche Comparable. La scelta più ovvia per un ListenableFuture è ListenableFutureTask, ma quella classe è final, quindi non è possibile creare istanze Comparable. La soluzione è creare un ListenableFutureTask e avvolgerlo in un SimpleForwardingListenableFuture che implementa Comparable.

---

Perché presumo hai a che fare con submit() piuttosto che execute()?

listeningDecorator(...).execute() non va a capo l'operazione di ingresso, come mostrato da questo test ho appena scritto:

public void testListeningDecorator_noWrapExecuteTask() { 
    ExecutorService delegate = mock(ExecutorService.class); 
    ListeningExecutorService service = listeningDecorator(delegate); 
    Runnable task = new Runnable() { 
    @Override 
    public void run() {} 
    }; 
    service.execute(task); 
    verify(delegate).execute(task); 
}