Como excluir seletivamente as mensagens de uma fila AMQP (RabbitMQ)?

Question

Gostaria de excluir seletivamente mensagens de uma fila AMQP sem sequer lê -las.

O cenário é o seguinte:

Enviar o lado deseja expirar as mensagens do tipo X com base no fato de que novas informações do tipo X chegaram. Como é muito provável que o assinante ainda não tenha consumido uma mensagem mais recente do tipo X, o editor deve excluir mensagens anteriores do tipo X e colocar uma mais recente na fila. Toda a operação deve ser transparente para o assinante - na verdade, ele deve usar algo tão simples quanto Stomp para obter as mensagens.

Como fazer isso usando o AMQP? Ou talvez seja mais conveniente em outro protocolo de mensagens?

Eu gostaria de evitar uma infraestrutura complicada. Todas as mensagens necessárias são tão simples quanto acima: uma fila, um assinante, um editor, mas o editor deve ter a capacidade de excluir as mensagens para um determinado critério.

O cliente do editor usará Ruby, mas na verdade eu lidaria com qualquer idioma assim que descobrir como fazê -lo no protocolo.

Answer 1

Atualmente, você não pode fazer isso no RabbitMQ (ou em geral, no AMQP) automaticamente. Mas aqui está uma solução alternativa fácil.

Digamos que você queira enviar três tipos de mensagens: XS, YS e ZS. Se eu entender sua pergunta corretamente, quando uma mensagem X chega, você deseja que o corretor esqueça todas as outras mensagens X que não foram entregues.

Isso é bastante fácil de fazer no RabbitMQ:

• O produtor declara três filas: x, y e z (eles estão automaticamente vinculados à troca padrão com seus nomes como teclas de roteamento, que é exatamente o que queremos),
• Ao publicar uma mensagem, o produtor primeiro elimina a fila relevante (então, se estiver publicando uma mensagem X, primeiro expurga a fila X); Isso remove efetivamente as mensagens desatualizadas,
• O consumidor simplesmente consome a partir da fila que deseja (x para x mensagens, y para y mensagens etc.); Do seu ponto de vista, ele só precisa fazer um básico para obter a próxima mensagem relevante.

Isso implica uma condição de corrida quando dois produtores enviam o mesmo tipo de mensagem ao mesmo tempo. O resultado é que é possível que a fila A tenha duas (ou mais) mensagens ao mesmo tempo, mas como o número de mensagens é fundamentado pelo número de produtores, e como as mensagens supérfluas são purgadas na próxima publicação , isso não deve ser um grande problema.

Para resumir, esta solução tem apenas uma etapa extra da solução ideal, a saber, purga da fila x antes de publicar uma mensagem do tipo X.

Se você precisar de ajuda para configurar essa configuração, o local perfeito para pedir conselhos é a lista de discussão RabbitMQ-Discuss.

Answer 2

Você não deseja uma fila de mensagens, deseja um banco de dados de valor-chave. Por exemplo, você pode usar o Redis ou Tokyo Tyrant para obter um banco de dados de valor-chave simples e acessível por rede. Ou apenas use um memcache.

Cada tipo de mensagem é uma chave. Quando você escreve uma nova mensagem com a mesma chave, ela substitui o valor anterior para que o leitor deste banco de dados nunca possa obter informações desatualizadas.

Nesse ponto, você só precisa de uma fila de mensagens para estabelecer a ordem em que as chaves devem ser lidas, se isso for importante. Caso contrário, basta digitalizar continuamente o banco de dados. Se você digitalizar continuamente o banco de dados, é melhor colocar o banco de dados perto dos leitores para reduzir o tráfego da rede.

Eu provavelmente faria algo assimkey: typecode value: lastUpdated, important data

Então eu enviaria mensagens que contêmtypecode, lastUpdated Dessa forma, o leitor pode comparar o LastUpdated para a chave com a que eles lêem pela última vez no banco de dados e ignoram a leitura, porque eles já estão atualizados.

Se você realmente precisa fazer isso com o AMQP, use o RabbitMQ e um tipo de troca personalizado, especificamente uma troca de cache de último valor. O código de exemplo está aqui https://github.com/squaremo/rabbitmq-lvc-plugin

Answer 3

Parece também funcionar do RabbitMQ Web-UI, se você quiser apenas remover as primeiras N mensagens da fila

• Selecione a fila da guia "Filas", role para baixo até a seção "Get Mensagens"
• Defina o parâmetro "Requeue = não" e número de mensagens que você deseja remover da fila
• Pressione o botão "Get Mensagens"

Answer 4

Esta pergunta tem alta visibilidade devido ao título. Passar pela descrição habita com um cenário mais específico. Portanto, para os usuários que desejam excluir a próxima mensagem (lembre -se de FIFO) da fila, você pode usar o Rabbitmqadmin e emitir o comando abaixo:

rabbitmqadmin get queue=queuename requeue=false count=1

Este comando está essencialmente consumindo a mensagem e não está fazendo nada. Um comando completo com sinalizador para receber o backup das mensagens pode parecer o abaixo. Certifique -se de adicionar outros parâmetros conforme sua exigência.

sudo python rabbitmqadmin -V virtualhostname -u user -p pass get queue=queuename requeue=false count=1 payload_file=~/origmsg