Mensajes normales - ApsaraMQ for RocketMQ - Centro de documentación de Alibaba Cloud

Los mensajes normales son el tipo de mensaje predeterminado en ApsaraMQ for RocketMQ. A diferencia de los mensajes ordenados, programados, diferidos y transaccionales, no poseen funciones especiales. Este tipo de mensaje admite comunicaciones asíncronas desacopladas entre productores y consumidores. Utilice mensajes normales cuando su aplicación requiera una entrega confiable, pero no necesite un orden estricto ni entregas temporizadas.

Cuándo usar mensajes normales

Los mensajes normales son adecuados para cualquier escenario que priorice la confiabilidad de la transmisión sobre el orden de procesamiento. Algunos ejemplos comunes incluyen el desacoplamiento de microservicios, la integración de datos y las arquitecturas basadas en eventos:

Desacoplamiento de microservicios: Un servicio ascendente encapsula la creación de pedidos y el pago como un mensaje normal independiente y lo envía al broker de ApsaraMQ for RocketMQ. Los servicios descendentes se suscriben de forma independiente y procesan el mensaje según su propia lógica. Cada mensaje es autónomo y no depende de otros mensajes.
Integración de datos: Un componente de instrumentación recopila registros de operación de aplicaciones frontend y los reenvía a ApsaraMQ for RocketMQ. Cada mensaje constituye un registro discreto de datos. El broker almacena y entrega estos registros a los sistemas de almacenamiento descendentes sin transformarlos. Las tareas posteriores las procesan las aplicaciones backend.

Ciclo de vida del mensaje

Un mensaje normal pasa por cinco etapas entre su creación y eliminación:

Lifecycle of a normal message

Etapa	Descripción
Inicializado	El productor construye el mensaje y lo prepara para el envío.
Listo	El broker acepta el mensaje. En este punto, el mensaje es visible y está disponible para los consumidores.
En tránsito	Un consumidor recupera el mensaje y comienza a procesarlo. Si el broker no recibe confirmación dentro de un periodo específico, retries delivery.
Confirmado	El consumidor confirma el resultado del consumo. El broker marca si el mensaje actual se consumió, pero no lo elimina de inmediato.
Eliminado	El mensaje se elimina cuando expira el periodo de retención o el espacio de almacenamiento se agota. La eliminación sigue una estrategia rotativa que borra primero los mensajes más antiguos del archivo físico. Para más detalles, consulte Message storage and cleanup.

De forma predeterminada, ApsaraMQ for RocketMQ conserva todos los mensajes incluso después de confirmarlos. Los consumidores pueden volver a consumir cualquier mensaje que aún no se haya eliminado.

Ejemplos de código (Java)

A continuación, los ejemplos utilizan el SDK de Apache RocketMQ 5.x para enviar y consumir mensajes normales. Cada ejemplo configura una conexión de cliente y ejecuta una sola operación. Reemplace los marcadores de posición con sus valores reales antes de ejecutar el código.

Para consultar la referencia completa del SDK y ejemplos en otros lenguajes, consulte Apache RocketMQ 5.x SDKs.

Antes de empezar

Prepare los siguientes recursos en la consola de ApsaraMQ for RocketMQ:

Recurso	Ubicación
Endpoint	Página Instance Details > pestaña Endpoints. Use el endpoint VPC para acceder desde instancias de Elastic Compute Service (ECS) en la misma nube privada virtual (VPC). Use el endpoint público para acceder a través de Internet (requiere habilitar la función de acceso a Internet).
Topic	Cree un topic con `MessageType` establecido en `Normal`. Los mensajes normales solo se pueden enviar a topics de este tipo.
Consumer group	Obligatorio para los ejemplos de consumidor. Cree uno en la consola.
Credenciales (solo endpoint público)	Nombre de usuario y contraseña de la página Access Control > pestaña Intelligent Authentication. No son necesarias para el acceso VPC, where el broker autentica automáticamente. Tampoco se requieren credenciales para instancias serverless a las que se accede mediante VPC con la función de autenticación gratuita en VPC habilitada.

Referencia de marcadores de posición

Marcador de posición	Descripción	Ejemplo
`<your-endpoint>`	Endpoint de la instancia	`rmq-cn-xxx.cn-hangzhou.rmq.aliyuncs.com:8080`
`<your-topic>`	Nombre del topic	`normal-topic-01`
`<your-consumer-group>`	Nombre del grupo de consumidores	`GID_normal_consumer`
`<your-username>`	Nombre de usuario de la instancia (solo endpoint público)	--
`<your-password>`	Contraseña de la instancia (solo endpoint público)	--

Código de ejemplo

Enviar un mensaje normal

package doc;

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 {
        // Specify the instance endpoint.
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // Uncomment the following line for public endpoint access:
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        ClientConfiguration configuration = builder.build();

        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        // Build a normal message with a key and tag for filtering and tracing.
        Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setKeys("messageKey")
                .setTag("messageTag")
                .setBody("messageBody".getBytes())
                .build();

        try {
            SendReceipt sendReceipt = producer.send(message);
            System.out.println("Send success. Message ID: " + sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

Salida esperada:

Send success. Message ID: 01BE0A3600F5762D0455025C2D1A0000

Consumir con un push consumer

Un push consumer recibe mensajes mediante una devolución de llamada MessageListener. El broker envía los mensajes al consumidor automáticamente.

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

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // Uncomment the following line for public endpoint access:
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        ClientConfiguration clientConfiguration = builder.build();

        // Subscribe to all messages in the topic (tag = "*").
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    System.out.println("Received message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();

        // Keep the process running to continue receiving messages.
        Thread.sleep(Long.MAX_VALUE);

        // Close the consumer when it is no longer needed.
        // pushConsumer.close();
    }
}

Salida esperada:

Received message: MessageView{messageId=01BE0A3600F5762D0455025C2D1A0000, topic=<your-topic>, ...}

Consumir con un simple consumer

Un simple consumer extrae mensajes explícitamente mediante receive() y confirma cada mensaje con ack(). Este modo le otorga control total sobre el ritmo de consumo.

import org.apache.rocketmq.client.apis.*;
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.MessageId;
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);

    public static void main(String[] args) throws ClientException, IOException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        // Uncomment the following line for public endpoint access:
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("<your-username>", "<your-password>"));
        ClientConfiguration clientConfiguration = builder.build();

        Duration awaitDuration = Duration.ofSeconds(10);
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(awaitDuration)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();

        int maxMessageNum = 16;
        Duration invisibleDuration = Duration.ofSeconds(10);

        // Pull and acknowledge messages in a loop.
        // For real-time consumption, use multiple threads to pull concurrently.
        while (true) {
            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
            messages.forEach(messageView -> {
                System.out.println("Received message: " + messageView);
            });
            for (MessageView message : messages) {
                final MessageId messageId = message.getMessageId();
                try {
                    consumer.ack(message);
                    System.out.println("Acked message: " + messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }

        // Close the consumer when it is no longer needed.
        // consumer.close();
    }
}

Salida esperada:

Received message: MessageView{messageId=01BE0A3600F5762D0455025C2D1A0000, topic=<your-topic>, ...}
Acked message: 01BE0A3600F5762D0455025C2D1A0000

Mejores prácticas

Establezca una clave de mensaje única globalmente para cada mensaje

Asigne un identificador específico del negocio, como un ID de pedido o un ID de usuario, como clave del mensaje. Esto le permite consultar mensajes individuales y sus trazas en la consola de ApsaraMQ for RocketMQ.

Message message = provider.newMessageBuilder()
        .setTopic(topic)
        .setKeys("order-12345")   // Use a unique business identifier.
        .setTag("payment")
        .setBody(payload)
        .build();

Elija el modo de consumidor adecuado

Modo	Funcionamiento	Recomendado para
Push consumer	El broker envía mensajes a una devolución de llamada `MessageListener`.	Procesamiento de baja latencia y basado en eventos, where el consumidor maneja cada mensaje al llegar.
Simple consumer	La aplicación llama a `receive()` para extraer un lote y luego ejecuta `ack()` en cada mensaje tras procesarlo.	Cargas de trabajo que requieren control de flujo manual o procesamiento por lotes.