Multiplexing RabbitMQ message

Question

Por exemplo, eu tenho 4 fontes que publicam meterics. Eu gostaria de multiplexar / mesclar todas essas mensagens em uma fila / troca

--------+----+----+----+----+        -------+---------+----+---------+---------+
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

Existe uma maneira "nativa" de fazer isso no RabbitMQ ou devo escrever meu próprio consumidor / editor?

EDITAR 1

Algum exemplo para esclarecer, digamos que depois de algum tempo eu tenha

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

E depois

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

E depois

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

O pedido de consumo de resultado será: 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 então X, Y, Z, W então X, Y, Z, W então X, Y, Z ... X, Z ...

Dessa forma, mesmo que uma fonte esteja "enviando spam", todas as outras mensagens de outras fontes têm chance de ser consumidas.

Por razões técnicas / financeiras, preciso consumir apenas 1 mensagem por vez.

O consumidor é muito mais lento do que os produtores, mas os produtores publicam muito, mas ocasionalmente.

Se cada fonte publicada em uma bolsa vinculada à mesma fila, o resultado pode ser XXXXXXXXXXXXXX YYYYYYYY ZZZZZZZZZZZ WWWou XXXXX Y XXXXX YYY XXX YYYY ZZZZZZZZZZZ WWW(dependendo da taxa de publicação de cada fonte)

Answer 1

Acho que o que você deseja pode ser alcançado simplesmente executando um único script que se inscreve em todas as filas.

O principal requisito é usar um único encadeamento de aplicativo para tratar todas as mensagens, independentemente da fila de onde elas chegam. A aparência disso irá variar dependendo de qual linguagem e biblioteca cliente você está usando - se você estiver usando PHP, você realmente terá que sair do seu caminho para não ter um thread único, mas talvez haja algumas bibliotecas cliente que assume que cada retorno de chamada está em um thread de trabalho separado e você precisará de algum recurso compartilhado para que eles bloqueiem.

Em termos do lado RabbitMQ real das coisas, você precisará:

• registrar uma assinatura para o servidor enviar mensagens para você, com basic.consume; isto é geralmente recomendado sobre explicitamente polling com basic.getqualquer maneira
• use um único "canal" para todas as basic.consumechamadas
• use confirmações manuais para que as mensagens permaneçam na fila até que seu processo seja concluído
• definir um limite de pré-busca por fila de 1 combasic.qos

Se você tiver 4 filas, A, B, C e D, que têm quantidades variáveis ​​de mensagens quando você inicia o consumidor:

1. Quando você se inscrever pela primeira vez, o limite de pré-busca significará que uma mensagem de cada fila será enviada ao canal; chame-os de A1, B1, C1 e D1
2. A biblioteca cliente gerará um evento assíncrono em seu aplicativo para cada um deles, por sua vez
3. Seu único thread de trabalho tratará do primeiro desses eventos e começará a processar a mensagem A1
4. Até que você reconheça manualmente essa mensagem, nenhuma outra mensagem pode chegar
5. Depois de reconhecer a primeira mensagem (A1), uma nova mensagem pode ser pré-buscada dessa fila (A2)
6. Enquanto isso, sua thread de trabalho irá desbloquear e lidar com o próximo evento que já foi gerado, para a mensagem B1
7. Apenas depois de processar os eventos pendentes para B1, C1 e D1, o thread de trabalho verá o evento para a mensagem A2
8. Enquanto as filas tiverem mensagens em espera, elas serão processadas em rodízio. Mesmo se todas as filas, exceto uma, ficarem vazias, elas retornarão à rotação assim que uma mensagem chegar, porque apenas uma mensagem da fila ocupada terá sido pré-buscada, o resto estará apenas esperando no servidor RabbitMQ.