Implementazione Java "Queed Queue" per produttori veloci, consumatori lenti

Question

Ho uno scenario produttore-consumatore in cui i produttori producono molto più velocemente di quanto i consumatori possano consumare. In generale, la soluzione consiste nel bloccare i produttori poiché uno scenario produttore/consumatore funziona alla stessa velocità del componente più lento. La limitazione o il blocco dei produttori è non una buona soluzione perché la nostra applicazione offre abbastanza tempo per i consumatori di recuperare il ritardo.

Ecco un diagramma che rappresenta una "fase" pieno nella nostra applicazione contro uno scenario più comune:

 Our Application     Common Scenario 
2N +--------+--------+ 
    |PPPPPPPP|oooooooo|           P = Producer 
    |PPPPPPPP|oooooooo|           C = Consumer 
    N +--------+--------+  N +--------+--------+--------+  o = Other Work 
    |CPCPCPCP|CCCCCCCC|  |CPCPCPCP|CPCPCPCP|oooooooo|  N = number of tasks 
    |CPCPCPCP|CCCCCCCC|  |CPCPCPCP|CPCPCPCP|oooooooo| 
    -------------------  ---------------------------- 
    0  T/2  T  0  T/2  T  3T/2 

L'idea è quella di massimizzare il throughput non inibendo i produttori.

I dati su cui operano i nostri compiti sono facilmente serializzabili, quindi ho in programma di implementare una soluzione di filesystem per svuotare tutte le attività che non possono essere immediatamente soddisfatte.

Sto utilizzando Java ThreadPoolExecutor con un BlockingQueue con una capacità massima per garantire che non si esaurisca la memoria. Il problema è nell'implementare una tale coda "a livelli", in cui le attività che possono essere accodate in memoria vengono eseguite immediatamente, altrimenti i dati vengono accodati sul disco.

mi è venuta in mente due possibili soluzioni:

1. implementare una BlockingQueue da zero, usando l'implementazione LinkedBlockingQueue o ArrayBlockingQueue come riferimento. Questo può essere semplice come copiare l'implementazione nella libreria standard e aggiungere la lettura/scrittura del filesystem.
2. continuare a utilizzare un BlockingQueue implementazione standard, attuare un separato FilesystemQueue per memorizzare i miei dati, e l'utilizzo di uno o più thread per annullare l'accodamento file, creare Runnable s e accodare utilizzando il ThreadPoolExecutor.

Sono entrambi ragionevoli e c'è un approccio migliore?

Answer 1

La prima opzione è aumento la disposizione mucchio spazio formato, come suggerito da Dimitar Dimitrov, utilizzando il flag memoria -Xmx, ad esempio java -Xmx2048m

Da Oracle's Documentation: nota che la JVM utilizza più memoria del solo heap. Ad esempio, i metodi Java , stack di thread e handle nativi sono allocati nella memoria separata dall'heap, nonché dalle strutture di dati interne JVM.

Anche qui è un diagramma di come java mucchio memoria è classificati.

---

Il seconda opzione è quello di utilizzare una libreria che implementa la funzionalità richiesta. A tale scopo è possibile utilizzare ashes-queue

Dalla panoramica del progetto: Questa è un'implementazione FIFO semplice in Java che ha il supporto persistente. Cioè, se la coda è piena, i messaggi traboccanti rimarranno persi e quando saranno disponibili gli slot , verranno rimessi in memoria.

---

L'terza opzione è quella dicreare una propria implementazione. Per questo, è possibile visualizzare l'anteprima this thread che ti guida a tale scopo.

I tuoi suggerimenti sono inclusi in questa terza opzione. Entrambi sono ragionevoli. Dal punto di vista dell'implementazione, dovresti scegliere la prima opzione in quanto garantisce un'implementazione più semplice e un design pulito.

Answer 2

Sembra la situazione ideale per utilizzare una coda JMS, piuttosto che un file system.

Invece di utilizzare una coda di blocco, inviare i messaggi su una coda JMS persistente. Si può ancora provare l'approccio a più livelli, combinando una coda JMS in parallelo con uno BlockingQueue, postando nella coda JMS quando BlockingQueue è pieno, ma sono sicuro che l'approccio JMS puro funzionerà da solo.

Answer 3

Prima di partire per una soluzione più complessa, sei davvero sicuro che l'utilizzo di uno BlockingQueue limitato sia un affare per te? Potrebbe risultare che aumentare le dimensioni dell'heap e preallocare una capacità sufficientemente generosa è ancora buono per te. Ti permetterà di evitare la complessità e l'incertezza delle prestazioni, sul prezzo delle pause GC che si trovano nella tua zona di comfort.

Tuttavia, se il carico di lavoro è così sbilanciato da poter trarre vantaggio dal persistere di una quantità di messaggi che non possono essere contenuti nella memoria (rispetto a una comprovata coda di blocco MPMC), sembra che sia necessaria una versione più semplice e più piccola di ActiveMQ o il suo off-shoot Apollo. A seconda dell'applicazione, è possibile trovare utili le altre funzionalità di ActiveMQ, nel qual caso è possibile utilizzarlo direttamente. In caso contrario, è preferibile cercare nello spazio JMS, come suggerito da bowmore.