ApsaraMQ for RabbitMQ:Delayed messages

Casos de uso

• Tratamento de timeout de pedidos: Após criar um pedido, envie uma mensagem com atraso de 30 minutos. Ao receber a mensagem, o consumidor verifica se o pagamento foi concluído; caso contrário, encerra o pedido.

• Notificações agendadas: Acione tarefas em um momento futuro, como enviar um lembrete 24 horas antes de um compromisso.

• Nova tentativa com backoff: Reenfileire mensagens com falha usando atrasos progressivamente maiores (por exemplo, 1s, 2s, 4s, 8s) para evitar sobrecarga nos serviços downstream.

Como funciona

A solução nativa de mensagens com atraso não exige instalação de plug-ins nem topologia de dead-letter exchange:

1. O produtor publica uma mensagem com o cabeçalho delay em um exchange.

2. O exchange roteia a mensagem para uma fila.

3. Os consumidores recebem a mensagem apenas após o término do período de atraso.

Escolha uma abordagem

O ApsaraMQ for RabbitMQ suporta quatro abordagens de mensagens com atraso. Todas as abordagens do RabbitMQ open source funcionam sem alterações de código.

O bloqueio de cabeça de fila significa que as mensagens em uma fila devem expirar em ordem. Uma mensagem com TTL menor não pode expirar antes de uma mensagem anterior com TTL maior, o que pode causar atrasos inesperados.

Para novas implementações, use a abordagem nativa de mensagens com atraso. Ela possui a configuração mais simples e evita o bloqueio de cabeça de fila.

Observações de uso

Actual message TTL = min(message-level TTL, queue-level TTL) + delay period

Enviar mensagens com atraso (abordagem nativa)

Adicione uma chave delay aos cabeçalhos da mensagem com o valor do atraso em milissegundos como string.

A abordagem nativa usa a chave de cabeçalho delay com valor string (por exemplo, "5000" ), enquanto a abordagem de plug-in open source usa x-delay com valor inteiro (por exemplo, 5000 ).

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

// Establish a connection to the ApsaraMQ for RabbitMQ instance
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("<your-instance-endpoint>");
factory.setPort(5672);
factory.setUsername("<your-access-key>");
factory.setPassword("<your-secret-key>");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

// Set the delay to 5000 milliseconds (5 seconds)
Map<String, Object> headers = new HashMap<>();
headers.put("delay", "5000");

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .messageId(UUID.randomUUID().toString())
    .headers(headers)
    .build();

// Publish the delayed message to the target exchange
channel.basicPublish("<exchange-name>", "<routing-key>", props, "order timeout check".getBytes("UTF-8"));

Substitua os seguintes espaços reservados pelos valores reais:

Para exemplos de código em outras linguagens, consulte AMQP Demos.

Enviar mensagens com atraso (abordagem de plug-in open source)

O ApsaraMQ for RabbitMQ é compatível com o plug-in de exchange de mensagens com atraso open source. Nenhuma instalação de plug-in é necessária.

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-delayed-type", "direct");
channel.exchangeDeclare("ExchangeName", "x-delayed-message", true, false, args);

byte[] messageBodyBytes = "delayed payload".getBytes("UTF-8");

Map<String, Object> headers = new HashMap<String, Object>();
headers.put("x-delay", 5000);  // Delay for 5000 milliseconds (5 seconds)

AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers);
channel.basicPublish("ExchangeName", "", props.build(), messageBodyBytes);

Perguntas frequentes

Por que minha mensagem é entregue depois do atraso especificado?

Isso geralmente ocorre quando o consumidor usa basic.get (modo pull). O ApsaraMQ for RabbitMQ armazena mensagens em um cluster de nós distribuídos. No modo pull, cada chamada basic.get atinge apenas um nó; portanto, o consumidor pode não puxar imediatamente a mensagem do nó onde ela está armazenada. Alterne para basic.consume (modo push) para receber mensagens assim que o período de atraso terminar.

Próximos passos

• TTL de mensagem — Configure a expiração das mensagens.

• Limites do cluster — Verifique o período máximo de atraso para o tipo da sua instância.

• Crie um exchange — Configure exchanges no console do ApsaraMQ for RabbitMQ.

• AMQP Demos — Exemplos de código em várias linguagens de programação.