Principles and architecture of ordered messages - ApsaraMQ for RocketMQ

Los mensajes ordenados permiten a los consumidores procesar los mensajes exactamente en el mismo orden en que se enviaron. ApsaraMQ for RocketMQ utiliza grupos de mensajes para delimitar el orden: los mensajes dentro del mismo grupo siempre se entregan de forma secuencial, mientras que los mensajes de grupos distintos se procesan de manera independiente.

Casos de uso

Utilice mensajes ordenados cuando los sistemas posteriores deban procesar eventos en la secuencia exacta en que ocurrieron en el sistema anterior:

Emparejamiento de operaciones -- En el mercado de valores, cuando varias ofertas comparten el mismo precio, se aplica la regla de primera oferta, primera operación. El sistema de procesamiento de órdenes debe gestionar las ofertas en la secuencia exacta en que se colocaron.
Sincronización incremental de datos -- Al replicar cambios de base de datos (inserciones, actualizaciones, eliminaciones) a través de una cola de mensajes hacia un sistema de búsqueda, las operaciones deben reproducirse en el orden original. Una reproducción desordenada genera un estado inconsistente.

Funcionamiento del ordenamiento

El ordenamiento de mensajes de extremo a extremo consta de dos partes: orden de producción (lado del envío) y orden de consumo (lado de la recepción). Ambas condiciones deben cumplirse para garantizar un procesamiento estricto de tipo primero en entrar, primero en salir (FIFO).

Orden de producción

Para garantizar el orden de producción, cumpla las tres condiciones siguientes:

Mismo grupo de mensajes -- Establezca el mismo grupo de mensajes para todos aquellos que deban ordenarse entre sí. Los mensajes pertenecientes a grupos diferentes no tienen relación de orden.
Productor único -- Envíe todos los mensajes relacionados desde un solo productor. Aunque pertenezcan al mismo grupo de mensajes, los mensajes provenientes de distintos productores en sistemas diferentes carecen de un orden determinista.
Envío serial -- Envíe los mensajes de forma secuencial desde un solo hilo. El cliente del productor admite acceso multihilo, pero los mensajes enviados en paralelo desde hilos distintos no tienen un orden determinista.

Cuando se cumplen estas condiciones, los mensajes del mismo grupo se almacenan en la misma cola siguiendo el orden de envío.

Storage logic for ordered messages

Comportamiento de almacenamiento:

Los mensajes del mismo grupo se almacenan de forma secuencial en la misma cola.
Los mensajes de grupos diferentes pueden coexistir en la misma cola, pero su orden relativo no está garantizado.

En el diagrama anterior, el Grupo de mensajes 1 (G1-M1, G1-M2, G1-M3) y el Grupo de mensajes 4 (G4-M1, G4-M2) comparten la Cola 1. ApsaraMQ for RocketMQ garantiza el orden dentro de cada grupo, pero no entre grupos distintos.

Orden de consumo

Para garantizar el orden de consumo, dos mecanismos trabajan en conjunto:

Orden de entrega -- El protocolo del SDK y del servidor entrega los mensajes según el orden de almacenamiento. Siga estrictamente el patrón de recibir-procesar-confirmar. El procesamiento asíncrono puede romper la garantía de orden.

Importante
Con PushConsumer, los mensajes se entregan uno por uno siguiendo el orden de almacenamiento. Con SimpleConsumer, pueden llegar varios mensajes en una sola extracción; su aplicación debe procesarlos de forma secuencial y confirmar cada uno antes de llamar nuevamente a receive para el mismo grupo. Para más detalles, consulte Consumer types.
Reintentos limitados -- Cuando un mensaje ordenado falla tras alcanzar el número máximo de reintentos, se omite para desbloquear los mensajes subsiguientes. Elija una cantidad de reintentos que equilibre la fiabilidad con el riesgo de bloquear todo el grupo. Los reintentos dentro de un grupo de mensajes:
- No alteran la garantía de orden.
- No afectan a los mensajes de otros grupos.
- Sí bloquean los mensajes subsiguientes del mismo grupo hasta que se resuelva el mensaje actual.
Importante
Mientras se reintenta un mensaje ordenado fallido, los mensajes subsiguientes del mismo grupo permanecen bloqueados. Estos se entregan únicamente después de que el mensaje actual se procese correctamente o agote sus reintentos.

Combinaciones de orden de producción y consumo

El modo FIFO estricto requiere tanto orden de producción como de consumo. Sin embargo, no todos los consumidores necesitan entrega ordenada. Combine las opciones según sus requisitos de rendimiento y ordenamiento:

Orden de producción	Orden de consumo	Resultado
Grupo de mensajes configurado; envío serial	Ordenado	FIFO estricto dentro de cada grupo de mensajes
Grupo de mensajes configurado; envío serial	Concurrente	Orden cronológico con mejor esfuerzo; mayor rendimiento
Sin grupo de mensajes; envío desordenado	Ordenado	Orden estricto a nivel de cola (coincide con el orden de almacenamiento, no con el de envío)
Sin grupo de mensajes; envío desordenado	Concurrente	Orden cronológico con mejor esfuerzo

Ciclo de vida del mensaje

Un mensaje ordenado pasa por cinco estados:

Message lifecycle

Inicializado -- El productor construye el mensaje y prepara su envío.
Listo -- El mensaje llega al broker y se vuelve visible para los consumidores.
En tránsito -- Un consumidor recupera el mensaje y lo procesa. Si el broker no recibe confirmación dentro del tiempo de espera, reintenta la entrega. Para más detalles, consulte Reintento de consumo.
Confirmado -- El consumidor confirma el resultado. De forma predeterminada, ApsaraMQ for RocketMQ conserva todos los mensajes. El broker marca el mensaje como consumido, pero no lo elimina de inmediato.
Eliminado -- Cuando expira el periodo de retención o el almacenamiento escasea, el broker elimina los mensajes más antiguos de forma rotativa. Antes de la eliminación, los mensajes pueden volver a consumirse. Para más detalles, consulte Almacenamiento y limpieza de mensajes.

Importante

Un mensaje reintentado se trata como un mensaje nuevo. El ciclo de vida del mensaje original finaliza.
Mientras se reintenta un mensaje ordenado, los mensajes subsiguientes del mismo grupo permanecen bloqueados hasta que el mensaje actual se procese correctamente.

Límites

Los mensajes ordenados solo pueden enviarse a temas cuyo MessageType esté configurado como FIFO. El tipo de mensaje debe coincidir con el tipo del tema.

Importante

Si un grupo de consumidores está configurado para entrega ordenada, todos los mensajes consumidos por ese grupo se facturan como mensajes ordenados, independientemente de su tipo real. Si no requiere un orden estricto, configure el grupo para entrega concurrente y reduzca costos.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un tema con MessageType configurado como FIFO, creado en la consola de ApsaraMQ for RocketMQ
Un grupo de consumidores configurado para entrega ordenada (obligatorio solo si necesita consumo ordenado)
El RocketMQ 5.x gRPC SDK instalado en su proyecto

Importante

Los mensajes ordenados solo pueden enviarse a temas cuyo MessageType esté configurado como FIFO. Un grupo de consumidores que no esté configurado en modo de entrega ordenada entregará los mensajes de forma concurrente. Compruebe tanto el tipo de tema como el modo de entrega del grupo de consumidores antes de escribir cualquier código.

Optimización de la concurrencia de consumo

Con el RocketMQ 5.x gRPC SDK, PushConsumer puede distribuir mensajes de la misma MessageQueue a diferentes hilos según su grupo de mensajes. Cuanto mayor sea la variedad de valores en sus grupos de mensajes, mayor será la mejora en el rendimiento.

Versiones de SDK compatibles:

SDK	Versión mínima
Java	5.0.8
C++	5.0.3
Otros SDK	No compatible

Envío y consumo de mensajes ordenados

Los mensajes ordenados requieren un grupo de mensajes en cada llamada de envío. Diseñe los grupos de mensajes con la granularidad más fina que permita su negocio; por ejemplo, utilice un ID de pedido o un ID de usuario. Esto permite mantener el orden por entidad y maximizar el paralelismo entre entidades distintas.

Todos los ejemplos siguientes utilizan Java. Para ver muestras completas del SDK en otros lenguajes, consulte RocketMQ 5.x gRPC SDK.

Código de ejemplo

Enviar mensajes ordenados

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;

public class ProducerExample {
    public static void main(String[] args) throws ClientException {
        // Instance endpoint. Get it from the Endpoints tab on the instance details page.
        // Use the VPC endpoint for access from ECS over the internal network.
        // Use the public endpoint for access from a local machine or on-premises data center.
        String endpoints = "<your-endpoint>";
        // Topic name. Create the topic in the console first.
        String topic = "<your-fifo-topic>";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // For public network access, set the instance username and password.
        // Get them from the Intelligent Authentication tab on the Access Control page.
        // For internal network access from ECS, skip this -- the server authenticates via VPC.
        // For Serverless instances, set credentials for public access.
        // If authentication-free internal access is enabled, skip this for internal access.
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"));
        ClientConfiguration configuration = builder.build();
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();
        // Build an ordered message with a message group.
        Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setKeys("messageKey")         // Message key for lookup
                .setTag("messageTag")          // Tag for consumer-side filtering
                .setMessageGroup("fifoGroup001") // Message group -- keep values discrete to avoid hot spots
                .setBody("messageBody".getBytes())
                .build();
        try {
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

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

Marcador	Descripción	Ejemplo
`<your-endpoint>`	Endpoint de la instancia obtenido de la consola	`rmq-cn-xxx.cn-hangzhou.rmq.aliyuncs.com:8080`
`<your-fifo-topic>`	Nombre del tema FIFO	`order-events`
`<instance-username>`	Nombre de usuario de la instancia (solo acceso público)	`LTAI5tXxx`
`<instance-password>`	Contraseña de la instancia (solo acceso público)	`xXxXxXx`

Consumo con PushConsumer

PushConsumer entrega los mensajes ordenados uno por uno siguiendo el orden de almacenamiento. Configure el grupo de consumidores en modo de entrega ordenada desde la consola antes de iniciar el consumidor. De lo contrario, los mensajes se entregarán de forma concurrente.

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        // Instance endpoint. Get it from the Endpoints tab on the instance details page.
        String endpoints = "<your-endpoint>";
        // Topic and consumer group. Create both in the console first.
        String topic = "<your-fifo-topic>";
        String consumerGroup = "<your-consumer-group>";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // For public network access, set credentials.
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"));
        ClientConfiguration clientConfiguration = builder.build();
        // Subscribe to all tags.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    // Process the message and return the result.
                    System.out.println("Consume Message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();
        Thread.sleep(Long.MAX_VALUE);
        // Close the consumer when no longer needed.
        // pushConsumer.close();
    }
}

Consumo con SimpleConsumer

SimpleConsumer extrae mensajes en lotes. Procese los mensajes de cada lote de forma secuencial y confirme cada uno individualmente.

Importante

Para los mensajes del mismo grupo, si un mensaje anterior no se ha confirmado, llamar nuevamente a receive no devolverá los mensajes subsiguientes de ese grupo. Este mecanismo de bloqueo evita el procesamiento desordenado. Los mensajes de otros grupos no se ven afectados y pueden seguir recibirse de forma concurrente.

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    private SimpleConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException {
        // Instance endpoint. Get it from the Endpoints tab on the instance details page.
        String endpoints = "<your-endpoint>";
        // Topic and consumer group. Create both in the console first.
        String topic = "<your-fifo-topic>";
        String consumerGroup = "<your-consumer-group>";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // For public network access, set credentials.
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"));
        ClientConfiguration clientConfiguration = builder.build();

        Duration awaitDuration = Duration.ofSeconds(10);
        // Subscribe to all tags.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(awaitDuration)                    // Long polling timeout
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();
        int maxMessageNum = 16;
        Duration invisibleDuration = Duration.ofSeconds(10);       // Invisibility window for processing
        // Poll for messages in a loop. Use multiple threads for higher throughput.
        while (true) {
            final List<MessageView> messageViewList = consumer.receive(maxMessageNum, invisibleDuration);
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // Acknowledge each message after processing.
                try {
                    consumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        }
        // Close the consumer when no longer needed.
        // consumer.close();
    }
}

Solución de problemas de reintentos de consumo

Los reintentos de consumo ordenado para PushConsumer ocurren en el lado del cliente; el servidor no registra los detalles del reintento. Si el rastreo de un mensaje muestra un resultado de entrega failed, revise los registros del cliente consumidor.

Para consultar la ruta de los registros del cliente, véase Configuración de registros.

Busque las siguientes palabras clave en los registros del cliente:

Message listener raised an exception while consuming messages
Failed to consume fifo message finally, run out of attempt times

Mejores prácticas

Procese mensajes de forma serial, no por lotes

Consuma un mensaje a la vez. El consumo por lotes puede romper el orden.

Ejemplo: Los mensajes se envían en el orden 1 -> 2 -> 3 -> 4. Durante el consumo por lotes, los mensajes 2 y 3 se procesan juntos y fallan. Al reintentar, tanto el 2 como el 3 se vuelven a entregar, pero el mensaje 2 podría procesarse nuevamente después de que el mensaje 3 ya se haya procesado correctamente en otro intento, lo que resulta en un consumo desordenado.

Distribuya los grupos de mensajes para evitar puntos críticos

ApsaraMQ for RocketMQ utiliza el valor del grupo de mensajes para determinar en qué cola del servidor se almacena cada mensaje. Todos los mensajes del mismo grupo se enrutan a la misma cola. Concentrar demasiados mensajes en pocos grupos sobrecarga esas colas, crea puntos críticos de almacenamiento y limita la escalabilidad.

Utilice claves detalladas como grupos de mensajes; por ejemplo, IDs de pedidos o IDs de usuarios. Cuanto mayor sea la variedad de valores en sus grupos de mensajes, más uniforme será la distribución de mensajes entre las colas. Esto mantiene el orden de los mensajes para la misma entidad y distribuye la carga entre colas para entidades diferentes.