Esiste un modo per reinserire le attività nella coda executor

Question

Ho una serie di attività (ad esempio Runnable s) che devono essere eseguite da un Executor.
Ogni attività richiede che determinate condizioni siano valide per procedere. Sarei interessato a sapere se c'è un modo per configurare in qualche modo lo Executor per spostare le attività alla fine della coda e provare a eseguirle in un secondo momento quando la condizione sarebbe valida e l'attività sarà in grado di eseguirla e terminarla.
Così il comportamento potrebbe essere qualcosa di simile:

1. Thread-1 take attività dalla coda e run si chiama
2. All'interno run la condizione non è ancora valido
3. Task ferma e il compito Thread-1 posti alla fine della coda e ottiene l'attività successiva da eseguire
4. Successivamente in Thread-X (dal pool di thread) l'attività di prelievo di nuovo dalla condizione di coda è valida e l'attività è in corso eseguito

Answer 1

Creare prima l'esecutore.

Avete diverse possibilità.

Se suppongo che le tue attività implementino un'interfaccia semplice per interrogare il loro stato (qualcosa come un enum con 'NeedReschedule' o 'Completed'), quindi implementa un wrapper (implementando Runnable) per le tue attività che prenderanno il compito e l'esecutore come parametri di instanciazione. Questo wrapper eseguirà l'attività a cui è associato, verificherà il suo stato in seguito e, se necessario, riprogrammerà una copia di se stesso nell'esecutore prima di terminare.

In alternativa, è possibile utilizzare un meccanismo di execption per segnalare al wrapper che l'attività deve essere riprogrammata. Questa soluzione è più semplice, nel senso che non richiede una particolare interfaccia per il tuo compito, così che il semplice Runnable possa essere lanciato nel sistema senza problemi. Tuttavia, le eccezioni comportano più tempo di calcolo (costruzione di oggetti, traccia dello stack ecc.).

Ecco una possibile implementazione del wrapper utilizzando il meccanismo di segnalazione delle eccezioni. È necessario implementare la classe RescheduleException che estende Throwable, che può essere attivata dall'esecuzione eseguibile (non è necessaria un'interfaccia più specifica per l'attività in questa configurazione). Puoi anche usare un semplice RuntimeException come proposto in un'altra risposta, ma dovrai testare la stringa del messaggio per sapere se questa è l'eccezione che stai aspettando.

public class TaskWrapper implements Runnable { 

    private final ExecutorService executor; 
    private final Runnable task;  

    public TaskWrapper(ExecutorService e, Runnable t){ 
     executor = e; 
     task = t; 
    } 

@Override 
public void run() { 

    try { 
       task.run(); 
    } 
    catch (RescheduleException e) { 
     executor.execute(this); 
    } 
} 

Ecco un'applicazione molto semplice cottura fino a 200 operazioni avvolto in modo casuale chiedendo un riprogrammare.

class Task implements Runnable { 

    @Override 
    public void run(){ 
    if (Maths.random() > 0.5) 
     throw new RescheduleException(); 
    }  
} 


public class Main { 

public static void main(String[] args){ 

    ExecutorService executor = Executors.newFixedThreadPool(10); 

    int i = 200; 
      while(i--) 
     executor.execute(new TaskWrapper(executor, new Task()); 
} 
} 

si potrebbe anche avere un thread dedicato per monitorare gli altri thread risultati (utilizzando una coda di messaggi) e riprogrammare se necessario, ma si perde un thread, rispetto alle altre soluzioni.

Answer 2

In Java 6, il costruttore ThreadPoolExecutor prende uno BlockingQueue<Runnable>, che viene utilizzato per memorizzare le attività in coda. È possibile implementare una tale coda di blocco che sovrascrive lo poll() in modo che se viene effettuato un tentativo di rimuovere ed eseguire un lavoro "pronto", quindi poll procede normalmente.Altrimenti il ​​runnable è posto sul retro della coda e si tenta di eseguire nuovamente il polling, possibilmente dopo un breve timeout.

Answer 3

A meno che non sia necessario attendere, è possibile aggiungere un'attività ripetuta a un servizio ScheduledExecutor con un intervallo di polling appropriato che si annulla o si elimina dopo che è "valido" per l'esecuzione.

ScheduleExecutorService ses = ... 

ses.scheduleAtFixedRate(new Runnable() { 
    public void run() { 
     if (!isValid()) return; 
     preformTask(); 
     throw new RuntimeException("Last run"); 
    } 
}, PERIOD, PERIOD, TimeUnit.MILLI_SECONDS);