O blog da AWS

Implementando a priorização de mensagens com filas de quórum no Amazon MQ para RabbitMQ

Por Akhil Melakunta, Sr. Solutions Architect e Vinodh Kannan Sadayamuthu, Sr. WW Spec. SA.

As filas de quórum agora estão disponíveis no Amazon MQ para RabbitMQ a partir da versão 3.13. As filas de quórum são um tipo de fila replicada de First in, First out (FIFO) que usa o algoritmo de consenso Raft para manter a consistência dos dados. As filas de quórum na versão 3.13 do RabbitMQ carecem de um recurso fundamental em comparação com as filas clássicas: priorização de mensagens. No entanto, a versão 4.0 do RabbitMQ introduziu suporte para prioridade de mensagem, que se comporta de forma diferente das prioridades clássicas de mensagens de fila. A migração de aplicativos de filas clássicas com prioridade de mensagem para filas de quórum no Amazon MQ para RabbitMQ apresenta desafios para os clientes. Esta postagem descreve as diferentes abordagens para implementar a priorização de mensagens em filas de quórum no Amazon MQ para RabbitMQ.

O Amazon MQ é um serviço gerenciado de agente de mensagens para Apache ActiveMQ e RabbitMQ que simplifica a configuração e a operação de agentes de mensagens na AWS.

Por que a priorização de mensagens é importante

Os sistemas de mensagens modernos exigem que as mensagens sejam tratadas de forma diferente, dependendo da prioridade comercial. Algumas mensagens são mais urgentes ou críticas do que outras, e priorizá-las pode aumentar a eficiência e a capacidade de resposta dos aplicativos. A priorização de mensagens permite que determinadas mensagens sejam processadas antes de outras, alinhando-se às prioridades de negócios e ajudando a garantir que mensagens de alto valor ou urgentes recebam a atenção de que precisam.

A priorização de mensagens aborda desafios comerciais críticos em vários setores. Nas seguradoras, ele pode agilizar o processamento urgente de sinistros priorizando mensagens de alta prioridade em vez de atualizações rotineiras de apólices, reduzindo os tempos de liquidação. Os fabricantes automotivos podem garantir que os alertas críticos da linha de produção e as notificações de segurança tenham precedência sobre os dados de telemetria padrão, evitando tempos de inatividade dispendiosos. As concessionárias de energia podem priorizar alertas de estabilidade da rede em tempo real e notificações de interrupção, permitindo respostas mais rápidas a possíveis apagões. Ao implementar a prioridade de mensagens, os setores podem direcionar a atenção imediata para operações urgentes e, ao mesmo tempo, gerenciar com eficiência os processos de rotina na infraestrutura existente. Ao usar essa abordagem para transformar suas estratégias de comunicação, as organizações podem responder com mais rapidez e eficácia a eventos críticos.

Filas clássicas em comparação com filas de quórum: priorização de mensagens

Nesta seção, explore as diferenças fundamentais entre filas clássicas e filas de quórum quando se trata de recursos de priorização de mensagens. Examine como cada tipo de fila lida com a prioridade da mensagem, os recursos integrados disponíveis e as principais considerações.

Priorização de mensagens com filas clássicas

Nas filas clássicas, o RabbitMQ suporta prioridades de mensagens que variam de 1 a 255, sendo 1 a prioridade mais baixa e 255 a mais alta. No entanto, geralmente é recomendável usar um intervalo menor (por exemplo, 1-5) para melhorar o desempenho, porque o RabbitMQ precisa manter uma subfila interna para cada prioridade, de 1 até o valor máximo configurado para uma determinada fila. Uma faixa de prioridade mais ampla aumenta o custo de CPU e memória, o que pode afetar o desempenho do broker.

Comportamento prioritário da fila em filas clássicas:

  • As filas clássicas exigem o argumento x-max-priority para definir o número máximo de prioridades para uma determinada fila
  • Um procedimento envia uma mensagem com um valor de propriedade prioritário
  • Os consumidores não precisam de uma configuração especial para lidar com as prioridades
  • As mensagens com prioridade mais alta são entregues antes das mensagens com prioridade mais baixa
  • Dentro do mesmo nível de prioridade, as mensagens são entregues na ordem FIFO
  • Mensagens sem uma propriedade prioritária são tratadas como se sua prioridade fosse a mais baixa
  • Mensagens com uma prioridade maior que a máxima da fila são tratadas como se tivessem sido publicadas com a prioridade máxima

Exemplo de código Python para implementação de fila clássica com prioridade de mensagem:

#!/usr/bin/env python
import pika
import ssl
# Set up SSL context for secure connection
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
# Define credentials
credentials = pika.PlainCredentials('username', 'password') # Replace with actual credentials
# Set up connection parameters for Amazon MQ RabbitMQ broker
connection_parameters = pika.ConnectionParameters(
    host='b-example.mq.us-west-2.on.aws', # Replace with actual broker endpoint
    port=5671,
    credentials=credentials,
    ssl_options=pika.SSLOptions(context)
)
# Establish connection and create a channel
connection = pika.BlockingConnection(connection_parameters)
channel = connection.channel()
# Declare a direct exchange
# - direct exchanges route messages based on routing key
channel.exchange_declare(
    exchange='priority_exchange',
    exchange_type='direct',
)
# Declare a priority queue
# - x-max-priority=5 sets maximum priority level (0-5)
# - x-queue-type=classic specifies classic queue implementation
channel.queue_declare(
    queue='classic_priority_queue',
    arguments={
        'x-max-priority': 5,
        'x-queue-type': "classic"
    }
)
# Bind queue to exchange with routing key
# - This connects the queue to the exchange
# - Messages sent to the exchange with matching routing key will be routed to this queue
channel.queue_bind(
    queue='classic_priority_queue',
    exchange='priority_exchange',
    routing_key='priority_queue'
)
# Publish messages with different priorities
# Low priority message (priority=1)
channel.basic_publish(
    exchange='priority_exchange',
    routing_key='priority_queue',
    body='Low priority message',
    properties=pika.BasicProperties(priority=1)
)
print(" [x] Sent 'Low priority message'")
# Medium priority message (priority=2)
channel.basic_publish(
    exchange='priority_exchange',
    routing_key='priority_queue',
    body='Medium priority message',
    properties=pika.BasicProperties(priority=2)
)
print(" [x] Sent 'Medium priority message'")
# High priority message (priority=5)
channel.basic_publish(
    exchange='priority_exchange',
    routing_key='priority_queue',
    body='High priority message',
    properties=pika.BasicProperties(priority=5)
)
print(" [x] Sent 'High priority message'")
# Close the connection
connection.close()

O código anterior demonstra a priorização de mensagens no RabbitMQ usando uma fila clássica com tratamento de prioridades integrado. A implementação se conecta a um broker RabbitMQ usando a biblioteca Python Pika e declara uma troca direta, uma fila clássica com um nível máximo de prioridade de 5. As mensagens são então publicadas nessa fila única com valores de prioridade atribuídos explicitamente (1 para baixa, 2 para média e 5 para alta prioridade). Quando os consumidores buscam mensagens dessa fila, o RabbitMQ entregará primeiro as mensagens de maior prioridade.

Priorização de mensagens com filas de quórum

Diferentemente das filas clássicas, as filas de quórum no Rabbit MQ 3.13 não oferecem suporte nativo à priorização de mensagens. No entanto, existem padrões eficazes que você pode implementar para obter prioridade de mensagem com filas de quórum.

Usando filas separadas para prioridades diferentes

Um método simples é criar várias filas de quórum, cada uma dedicada a diferentes níveis de prioridade. Por exemplo, você pode ter uma fila de alta prioridade e uma fila de baixa prioridade. Usando o RabbitMQ Exchange e vinculando mensagens de roteamento de chaves às filas apropriadas com base em sua prioridade, permitindo que o sistema processe mensagens de alta prioridade mais rapidamente, conforme mostrado na figura a seguir.

Exemplo para implementar o tratamento prioritário usando filas de quórum separadas:

#!/usr/bin/env python
import pika
import ssl
# Set up SSL context for secure connection
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
# Define credentials
credentials = pika.PlainCredentials('username', 'password') #Replace with actual credentials
# Set up connection parameters for Amazon MQ RabbitMQ broker
connection_parameters = pika.ConnectionParameters(
    host='b-example.mq.us-west-2.on.aws',
    port=5671,
    credentials=credentials,
    ssl_options=pika.SSLOptions(context)
)
# Establish connection and create a channel
connection = pika.BlockingConnection(connection_parameters)
channel = connection.channel()
# Declare a direct exchange
# - Direct exchanges route messages based on routing key
channel.exchange_declare(
    exchange='priority_exchange_qq',
    exchange_type='direct'
)
# Create separate quorum queues for different priority levels
# Low priority queue
channel.queue_declare(
    queue='low_priority_queue',
    durable=True,
    arguments={
        'x-queue-type': "quorum" 
    }
)
# Bind the low priority queue to the exchange with a specific routing key
# - This creates a rule that messages sent to 'priority_exchange' with routing_key='low_priority_1'
# - will be routed to the 'low_priority_queue'
channel.queue_bind(
    queue='low_priority_queue',
    exchange='priority_exchange_qq',
    routing_key='low_priority_1'
)
# Medium priority queue
channel.queue_declare(
    queue='medium_priority_queue',
    durable=True,
    arguments={
        'x-queue-type': "quorum" 
    }
)
# Bind the medium priority queue to the exchange with a specific routing key
# - Messages with routing_key='medium_priority_2' will be directed to the 'medium_priority_queue'
channel.queue_bind(
    queue='medium_priority_queue',
    exchange='priority_exchange_qq',
    routing_key='medium_priority_2'
)
# High priority queue
channel.queue_declare(
    queue='high_priority_queue',
    durable=True,
    arguments={
        'x-queue-type': "quorum" 
    }
)
# Bind the high priority queue to the exchange with a specific routing key
# - Messages with routing_key='high_priority_2' will be directed to the 'high_priority_queue'
channel.queue_bind(
    queue='high_priority_queue',
    exchange='priority_exchange_qq',
    routing_key='high_priority_5'
)
# Publish messages to different priority queues
print(" [x] Publishing messages to different priority queues")
# Low priority message
channel.basic_publish(
    exchange='priority_exchange_qq',  
    routing_key='low_priority_1',
    body='Low priority message'
)
print(" [x] Sent 'Low priority message'")
# Medium priority message
channel.basic_publish(
    exchange='priority_exchange_qq', 
    routing_key='medium_priority_2',
    body='Medium priority message'
)
print(" [x] Sent 'Medium priority message'")
# High priority message
channel.basic_publish(
    exchange='priority_exchange_qq', 
    routing_key='high_priority_5',
    body='High priority message'
)
print(" [x] Sent 'High priority message'")
# Close the connection
connection.close()
print(" [x] Connection closed")

O código anterior demonstra uma abordagem de priorização de mensagens no RabbitMQ usando filas de quórum separadas para diferentes níveis de prioridade (baixo, médio e alto). A implementação usa a biblioteca Python Pika para se conectar a um servidor RabbitMQ, uma troca direta e três filas de quorum separadas para diferentes níveis de prioridade, além de publicar mensagens em chaves de roteamento diferentes com prioridades diferentes.

Lógica de prioridade personalizada para consumidores

Implemente uma lógica personalizada em seu aplicativo para lidar com mensagens com base em sua prioridade. Por exemplo, você pode usar cabeçalhos ou metadados para determinar a prioridade de uma mensagem e, em seguida, usar essas informações para rotear mensagens para filas diferentes ou processá-las em uma ordem específica.

As filas de prioridade mais alta devem usar mais consumidores ou consumidores com mais recursos alocados para processar mensagens mais rapidamente do que as filas de prioridade mais baixa. Use o método basic.qos (prefetch)no modo de reconhecimento manual em seus consumidores para limitar o número de mensagens que podem ser enviadas para entrega a qualquer momento e permitir que as mensagens sejam priorizadas. basic.qos é um valor que um consumidor define ao se conectar a uma fila. Ele indica quantas mensagens o consumidor pode manipular ao mesmo tempo. Esse método é mostrado na figura a seguir.

Observação: essa solução implementa a prioridade das mensagens com base no melhor esforço. Existe a possibilidade de que mensagens de baixa e média prioridade sejam processadas antes das mensagens de alta prioridade.

Conclusão

A priorização de mensagens nos brokers  RabbitMQ no Amazon MQ tem considerações diferentes para filas clássicas e de quórum. O uso de filas de quórum requer uma abordagem cuidadosa devido à falta de suporte nativo para a priorização de mensagens no RabbitMQ. Ao empregar filas separadas e lógica personalizada, você pode obter uma priorização efetiva e, ao mesmo tempo, manter a alta disponibilidade e a consistência que as filas de quórum oferecem. Adote essas estratégias para otimizar sua infraestrutura de mensagens, aprimorar a capacidade de resposta do aplicativo e garantir que as mensagens críticas sejam processadas em tempo hábil.

Recomendamos que você adote as filas de quórum como o tipo preferido de fila replicada nos brokers RabbitMQ 3.13. Para obter mais detalhes, consulte a documentação do Amazon MQ. Para obter mais informações, consulte filas de quórum.

Para saber mais, consulte Amazon MQ for Rabbit MQ.

Este conteúdo foi traduzido da postagem original do blog, que pode ser encontrada aqui.

Biografia dos Autores

Akhil Melakunta, Sr. Solutions Architect
Vinodh Kannan Sadayamuthu, Sr. WW Spec. SA

Biografia do Tradutor

Rodrigo Peres é Arquiteto de Soluções na AWS, com mais de 20 anos de experiência trabalhando com arquitetura de soluções, desenvolvimento de sistemas e modernização de sistemas legados.

Biografia do Revisor

Daniel Abib é arquiteto de soluções sênior na AWS, com mais de 25 anos trabalhando com gerenciamento de projetos, arquiteturas de soluções escaláveis, desenvolvimento de sistemas e CI/CD, microsserviços, arquitetura Serverless & Containers e segurança. Ele trabalha apoiando clientes corporativos, ajudando-os em sua jornada para a nuvem.

https://www.linkedin.com/in/danielabib/