Messaggio RabbitMQ multiplexing

Question

Ad esempio, ho 4 fonti che pubblicano meterics. Vorrei multiplex / unire tutti i messaggi di tesi in una coda / scambio

--------+----+----+----+----+        -------+---------+----+---------+---------+
Source1 | M1 | M2 | M3 |    |   =>   Result | M1 | M4 | M2 | M3 | M6 | M5 | M7 |
Source2 | M4 |    |    | M5 |
Source3 |    |    | M6 |    |
Source4 |    |    |    | M7 |

For each queue:
 * Read one message
 * Publish message to the Result queue

Esiste un modo "nativo" per farlo in RabbitMQ o dovrei scrivere il mio Consumer / Publisher?

MODIFICA 1

Qualche esempio per chiarire, diciamo dopo un po 'di tempo che ho

     Processing "window"
        +-+
Source1 |X|XXXXXXXXXXXXX
Source2 |Y|YYYYYYY
Source3 |Z|ZZZZZZZZZZ
Source4 |W|WW
        +-+

E poi più tardi

     Processing "window"
           +-+
Source1 XXX|X|XXXXXXXXXX
Source2 YYY|Y|YYYY
Source3 ZZZ|Z|ZZZZZZZ
Source4 WWW| |
           +-+

E poi più tardi

     Processing "window"
                 +-+
Source1 XXXXXXXXX|X|XXXX
Source2 YYYYYYYY | |
Source3 ZZZZZZZZZ|Z|Z
Source4 WWW      | |
                 +-+

Il risultato che consuma l'ordine sarà: X Y Z W X Y Z W X Y Z W X Y Z X Y Z X Y Z X Y Z X Y Z X Z X Z X Z X X X

X, Y, Z, W poi X, Y, Z, W poi X, Y, Z, W poi X, Y, Z ... X, Z ...

In questo modo, anche se una fonte "invia spam", tutti gli altri messaggi provenienti da altre fonti hanno la possibilità di essere consumati.

Per motivi tecnico / finanziari devo consumare solo 1 messaggio alla volta.

Il consumatore è molto più lento dei produttori, ma i produttori pubblicano molto ma occasionalmente.

Se ogni sorgente è stata pubblicata in uno scambio associato alla stessa coda, il risultato potrebbe essere XXXXXXXXXXXXXX YYYYYYYY ZZZZZZZZZZZ WWWo XXXXX Y XXXXX YYY XXX YYYY ZZZZZZZZZZZ WWW(a seconda del tasso di pubblicazione di ciascuna sorgente)

Answer 1

Penso che quello che vuoi può essere ottenuto semplicemente eseguendo un singolo script che si iscrive a tutte le code.

Il requisito fondamentale è utilizzare un singolo thread dell'applicazione per gestire tutti i messaggi, indipendentemente dalla coda da cui arrivano. L'aspetto varierà a seconda della lingua e della libreria client che stai utilizzando: se stai utilizzando PHP, dovresti davvero fare di tutto per non essere a thread singolo, ma forse ci sono alcune librerie client che presuppone che ogni callback si trovi su un thread di lavoro separato e avrai bisogno di una risorsa condivisa per bloccarli.

In termini di aspetto effettivo di RabbitMQ, dovrai:

• registrare un abbonamento per il server per inviare messaggi all'utente, con basic.consume; questo è generalmente consigliato rispetto al polling esplicito con basic.getanyway
• utilizzare un unico "canale" per tutte le basic.consumechiamate
• utilizzare riconoscimenti manuali in modo che i messaggi rimangano nella coda fino al termine del processo
• imposta un limite di prefetch per coda pari a 1 conbasic.qos

Se hai 4 code, A, B, C e D, che hanno quantità variabili di messaggi quando avvii il consumatore:

1. Quando ti iscrivi per la prima volta, il limite di precaricamento significherà che un messaggio da ciascuna coda verrà inviato al canale; chiamali A1, B1, C1 e D1
2. La libreria client genererà un evento asincrono nell'applicazione per ciascuno di questi a sua volta
3. Il tuo singolo thread di lavoro gestirà il primo di questi eventi e inizierà a elaborare il messaggio A1
4. Fino a quando non riconosci manualmente quel messaggio, nessun altro messaggio può arrivare
5. Dopo aver riconosciuto il primo messaggio (A1), è possibile prelevare un nuovo messaggio da quella coda (A2)
6. Nel frattempo, il tuo thread di lavoro si sbloccherà e gestirà il prossimo evento che è già stato generato, per il messaggio B1
7. Solo dopo aver elaborato gli eventi in sospeso per B1, C1 e D1, il thread di lavoro vedrà l'evento per il messaggio A2
8. Finché le code hanno messaggi in attesa, verranno elaborati in modo round-robin. Anche se tutte le code tranne una diventano vuote, torneranno in rotazione non appena arriva un messaggio, perché solo un messaggio dalla coda occupata sarà stato pre-recuperato, il resto sarà solo in attesa sul server RabbitMQ.