ApsaraMQ for RabbitMQ:Delayed messages

Casos de uso

• Gestión de tiempos de espera en pedidos: Tras crear un pedido, envíe un mensaje diferido con un retraso de 30 minutos. Cuando el consumidor recibe el mensaje, comprueba si se completó el pago; de lo contrario, cierra el pedido.

• Notificaciones programadas: Active tareas en un momento futuro, como enviar un recordatorio 24 horas antes de una cita.

• Reintento con retroceso progresivo: Vuelva a poner en cola los mensajes con errores con retrasos cada vez mayores (por ejemplo, 1 s, 2 s, 4 s, 8 s) para evitar saturar los servicios posteriores.

Funcionamiento

La solución nativa de mensajes diferidos no requiere instalar complementos ni configurar una topología de intercambio de letras muertas:

1. Un productor publica un mensaje con un encabezado delay en un intercambio.

2. El intercambio enruta el mensaje a una cola.

3. Los consumidores reciben el mensaje solo después de que finaliza el periodo de retraso.

Elección de un método

ApsaraMQ for RabbitMQ admite cuatro métodos de mensajes diferidos. Todos los métodos de RabbitMQ de código abierto funcionan sin necesidad de modificar el código.

El bloqueo de cabecera de línea significa que los mensajes de una cola deben caducar en orden. Un mensaje con un TTL más corto no puede caducar antes que un mensaje anterior con un TTL más largo, lo cual puede provocar retrasos inesperados.

Para nuevas implementaciones, utilice el método de mensaje diferido nativo, ya que tiene la configuración más sencilla y evita el bloqueo de cabecera de línea.

Notas de uso

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

Envío de mensajes diferidos (método nativo)

Agregue una clave delay a los encabezados del mensaje con el valor de retraso en milisegundos como cadena.

El método nativo utiliza la clave de encabezado delay con un valor de tipo cadena (por ejemplo, "5000" ), mientras que el método del complemento de código abierto usa x-delay con un valor entero (por ejemplo, 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"));

Reemplace los siguientes marcadores de posición con los valores reales:

Para ver código de ejemplo en otros lenguajes, consulte AMQP Demos.

Envío de mensajes diferidos (método del complemento de código abierto)

ApsaraMQ for RabbitMQ es compatible con el complemento de intercambio de mensajes diferidos de código abierto. No es necesario instalar ningún complemento.

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);

Preguntas frecuentes

¿Por qué mi mensaje se entrega después del retraso especificado?

Esto suele ocurrir cuando el consumidor utiliza basic.get (modo pull). ApsaraMQ for RabbitMQ almacena mensajes en un clúster de nodos distribuidos. En modo pull, cada llamada a basic.get llega solo a un nodo, por lo que es posible que el consumidor no obtenga inmediatamente el mensaje del nodo donde se almacena. Cambie a basic.consume (modo push) para recibir los mensajes tan pronto como finalice el periodo de retraso.

Pasos siguientes

• Message TTL: Configure la caducidad de los mensajes.

• Cluster limits: Consulte el periodo máximo de retraso para su tipo de instancia.

• Create an exchange: Configure intercambios en la consola de ApsaraMQ for RabbitMQ.

• AMQP Demos: Código de ejemplo en varios lenguajes de programación.