Utilizzo di Spring @Scheduled e @Async insieme

Question

Ecco il mio caso d'uso.

Un sistema legacy aggiorna una tabella code del database QUEUE.

voglio un processo ricorrente pianificato che - controlla il contenuto della CODA - se non ci sono righe della tabella non si blocca la riga e fa un certo lavoro - elimina la riga nella coda

Se il lavoro precedente è ancora in esecuzione, quindi verrà creata una nuova discussione per eseguire il lavoro. Voglio configurare il numero massimo di thread simultanei.

Sto usando Spring 3 e la mia soluzione attuale è quella di eseguire le seguenti operazioni (con un tasso fisso di 1 millisecondo per ottenere i thread per eseguire in pratica in modo continuo)

@Scheduled(fixedRate = 1) 
@Async 
public void doSchedule() throws InterruptedException { 
    log.debug("Start schedule"); 
    publishWorker.start(); 
    log.debug("End schedule"); 
} 

<task:executor id="workerExecutor" pool-size="4" /> 

Questo ha creato 4 thread di getto ed i fili condiviso correttamente il carico di lavoro dalla coda. Tuttavia, sembra che si stia verificando una perdita di memoria quando i thread impiegano molto tempo per essere completati.

java.util.concurrent.ThreadPoolExecutor @ 0xe097b8f0        |    80 | 373,410,496 |  89.74% 
|- java.util.concurrent.LinkedBlockingQueue @ 0xe097b940       |   48 | 373,410,136 |  89.74% 
| |- java.util.concurrent.LinkedBlockingQueue$Node @ 0xe25c9d68 

Così

1: Dovrei usare @Async e @Scheduled insieme?

2: In caso contrario, in quale altro modo posso utilizzare la molla per raggiungere i miei requisiti?

3: Come posso creare i nuovi thread solo quando gli altri thread sono occupati?

Grazie a tutti!

EDIT: Penso che la coda di posti di lavoro stava diventando infinitamente lungo ... Ora utilizzando

<task:executor id="workerExecutor" 
    pool-size="1-4" 
    queue-capacity="10" rejection-policy="DISCARD" /> 

Riferirà con risultati

Answer 1

//using a fixedRate of 1 millisecond to get the threads to run basically continuously 
@Scheduled(fixedRate = 1) 

Quando si utilizza @Scheduled verrà creato un nuovo thread e invocherà il metodo doSchedule alla frequenza fissa specificata a 1 millisecondo. Quando esegui la tua app puoi già vedere 4 thread in competizione per la tabella CODE e probabilmente un dead lock.

Indagare se c'è un deadlock prendendo discarica filo. http://helpx.adobe.com/cq/kb/TakeThreadDump.html

@L'annotazione di sincronizzazione non sarà utile.

Il modo migliore per implementare questo è creare la classe come thread implementando eseguibile e passando la classe a TaskExecutor con il numero di thread richiesto.

Using Spring threading and TaskExecutor, how do I know when a thread is finished?

Anche controllare il vostro disegno non sembra essere la manipolazione la sincronizzazione in modo corretto. Se un lavoro precedente è in esecuzione e in possesso di un blocco sulla riga, il lavoro successivo creato continuerà a vedere quella riga e attenderà l'acquisizione del blocco su quella particolare riga.

Answer 2

Si può provare

1. eseguire un programmatore con un secondo di ritardo, che si blocca & recuperare tutti i record code che non sono stati bloccati finora.
2. Per ogni record, chiamare un metodo Async, che elaborerà tale record & eliminarlo.
3. Il criterio di rifiuto dell'esecutore deve essere ABORT, in modo che lo scheduler possa sbloccare le QUEUE che non sono ancora state distribuite per l'elaborazione. In questo modo, lo scheduler può provare a elaborare nuovamente queste QUEUE nella prossima esecuzione.

Ovviamente, è necessario gestire lo scenario, in cui lo scheduler ha bloccato una CODA, ma il gestore non ha terminato l'elaborazione per qualsiasi motivo.

pseudo codice:

public class QueueScheduler { 
    @AutoWired 
    private QueueHandler queueHandler; 

    @Scheduled(fixedDelay = 1000) 
    public void doSchedule() throws InterruptedException { 
     log.debug("Start schedule"); 
     List<Long> queueIds = lockAndFetchAllUnlockedQueues(); 
     for (long id : queueIds) 
      queueHandler.process(id); 
     log.debug("End schedule"); 
    } 
} 

public class QueueHandler { 

    @Async 
    public void process(long queueId) { 
     // process the QUEUE & delete it from DB 
    } 
} 
<task:executor id="workerExecutor" pool-size="1-4" queue-capcity="10" 
    rejection-policy="ABORT"/>