Locks and batch fetch messages con RabbitMq

Question

Sto tentando di utilizzare RabbitMq in un modo più anticonvenzionale (anche se a questo punto posso selezionare qualsiasi altra implementazione di coda di messaggi, se necessario). Invece di lasciare i messaggi push di Rabbit ai miei consumatori, l'utente si connette a una coda e recupera un batch di messaggi N (durante il quale ne consuma alcuni e ne rifiuta alcuni), dopo di che salta a un'altra coda e così via. Questo è fatto per la ridondanza. Se alcuni consumatori si bloccano, tutti i messaggi sono garantiti per essere consumati da altri utenti.

Il problema è che ho più utenti e non voglio che competano sulla stessa coda. C'è un modo per garantire un blocco su una coda? In caso contrario, posso essere sicuro che se 2 utenti sono connessi alla stessa coda non leggono lo stesso messaggio? Le transazioni potrebbero aiutarmi in qualche modo, ma ho sentito parlare che verranno rimosse da RabbitMQ.

Anche altri suggerimenti architettonici sono ben accetti.

Grazie!

EDIT: Come sottolineato nel commento c'è un una particolarità nel modo in cui ho bisogno per elaborare i messaggi. Hanno senso solo in gruppi e c'è un'alta probabilità che i messaggi correlati siano raggruppati in una coda. Se ad esempio estraggo un lotto di 100 messaggi, è molto probabile che sarò in grado di fare qualcosa con i messaggi 1-3, 4-5,6-10 ecc. Se non riesco a trovare un gruppo per alcuni messaggi li riporterò in coda. WorkQueue non funzionerebbe perché diffonderebbe messaggi dallo stesso gruppo a più lavoratori che non saprebbero cosa fare con loro.

Answer 1

Avete dato un'occhiata a questo libro online gratuito su Enterprise Integration Patterns?

Sembra che tu abbia davvero bisogno di un flusso di lavoro in cui hai un componente batch prima che i messaggi arrivino ai tuoi lavoratori. Con RabbitMQ ci sono due modi per farlo. Utilizzare un tipo di scambio (e formato messaggio) che può eseguire il batching per l'utente o una coda e un operatore che ordina i batch e inserisce ciascun batch nella propria coda. Il batcher probabilmente dovrebbe anche inviare un messaggio "pronto per il batch" a una coda di controllo in modo che un operatore possa scoprire l'esistenza della nuova coda batch. Una volta che il batch è stato elaborato, il lavoratore può eliminare la coda batch.

Se si ha il controllo sul formato del messaggio, è possibile che RabbitMQ esegua il batch in modo implicito in un paio di modi. Con uno scambio di argomenti, è possibile assicurarsi che la chiave di instradamento su ciascun messaggio abbia il formato work.batchid.qualcosa e quindi un worker che impari dell'esistenza di batch xxyzz utilizzi una chiave di associazione come # .xxyzz. # Solo consumare quei messaggi Non è necessaria alcuna ripubblicazione.

L'altro modo è includere un ID batch in un'intestazione e utilizzare il tipo di scambio di intestazioni più recenti. Naturalmente puoi anche implementare i tuoi tipi di scambio personalizzati se sei disposto a scrivere una piccola quantità di codice Erlang.

Tuttavia, consiglio di controllare il libro perché offre una migliore panoramica dell'architettura di messaggistica rispetto al tipico concetto di coda di lavoro che la maggior parte delle persone inizia.

Answer 2

Fai prelevare i tuoi clienti da una sola coda. Saranno garantiti di non condividere messaggi (Rabbit arrotolerà i messaggi tra i consumatori attualmente connessi) ed è fortemente ottimizzato per il modello di utilizzo esatto.

È pronto all'uso, fuori dalla scatola. Nei documenti RabbitMQ è chiamato il modello Work Queue. Una coda, più utenti, nessuno dei quali condivide nulla. Sembra quello di cui hai bisogno.

Answer 3

È possibile impostare un conteggio prefetch di canale/livello consumer per consumare messaggi in batch. Per inviare nuovamente i messaggi, è necessario utilizzare il metodo AMQP basic.reject e tali messaggi possono essere scelti per essere riaccodati o inoltrati a una coda di messaggi non recapitati. Più utenti che tentano di estrarre messaggi dalla stessa coda non è un problema perché il metodo AMQP basic.get verrà sincronizzato per gestire utenti concorrenti.

https://groups.google.com/forum/#!topic/rabbitmq-users/hJ8f5du-GCA