Suscripción a mensajes - ApsaraMQ for RocketMQ - Centro de documentación de Alibaba Cloud

Suscríbase a mensajes mediante el SDK de cliente TCP para Java de ApsaraMQ for RocketMQ. Este tema aborda los modos de suscripción, los modos de entrega y ejemplos de código completos.

Modos de suscripción

ApsaraMQ for RocketMQ admite dos modos de suscripción: clúster y difusión.

Clúster (predeterminado)

En el modo de clúster, todos los consumidores del mismo grupo comparten la carga de mensajes por igual. Cada mensaje se entrega a un solo consumidor.

Por ejemplo, si un tema contiene 9 mensajes y el grupo tiene 3 consumidores, cada consumidor recibe 3 mensajes.

// Clustering is the default mode. To set it explicitly:
properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);

Difusión

En el modo de difusión, cada consumidor del grupo recibe una copia completa de cada mensaje.

Por ejemplo, si un tema contiene 9 mensajes y el grupo tiene 3 consumidores, cada consumidor recibe los 9 mensajes.

// Set the subscription mode to broadcasting.
properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

Nota

Todas las instancias de consumidor del mismo ID de grupo deben usar consistent subscriptions.
El modo de difusión no admite el envío ni la recepción de mensajes ordenados, el seguimiento del progreso de consumo ni el restablecimiento de desplazamientos del consumidor. Para obtener más información, consulte Clustering consumption and broadcasting consumption.

Modos de entrega

ApsaraMQ for RocketMQ admite dos modos de entrega: push y pull.

Modo	Descripción	Recomendado para
Push	Los mensajes se envían desde ApsaraMQ for RocketMQ a los consumidores. También admite batch consumption.	La mayoría de los casos de uso
Pull	Los consumidores obtienen mensajes de ApsaraMQ for RocketMQ, lo que ofrece más opciones en la recepción y mayor libertad en la extracción de mensajes.	Casos de uso que requieren control detallado sobre la obtención de mensajes

Importante

Los consumidores pull requieren una instancia Enterprise Platinum Edition.
Los consumidores pull solo pueden conectarse a través de una Virtual Private Cloud (VPC).

Para consultar la lista completa de métodos y parámetros del consumidor pull, consulte Methods and parameters.

Código de ejemplo

A continuación se presentan ejemplos que muestran el consumo push, push por lotes y pull. Para ver más ejemplos, consulte el repositorio de código de ApsaraMQ for RocketMQ en GitHub.

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

Marcador de posición	Descripción	Dónde encontrarlo
`<your-group-id>`	ID del grupo de consumidores	Consola de ApsaraMQ for RocketMQ
`<your-tcp-endpoint>`	Endpoint TCP	Sección TCP Endpoint en la página Instance Details de la consola

Todos los ejemplos leen las credenciales de AccessKey de las variables de entorno ALIBABA_CLOUD_ACCESS_KEY_ID y ALIBABA_CLOUD_ACCESS_KEY_SECRET.

Consumidor push

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ConsumerTest {
   public static void main(String[] args) {
       Properties properties = new Properties();
       // The consumer group ID created in the ApsaraMQ for RocketMQ console.
       properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
       // Read AccessKey credentials from environment variables.
       properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
       properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
       // The TCP endpoint from the Instance Details page in the console.
       properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
       // Clustering mode (default). Uncomment the line below to switch to broadcasting.
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

       Consumer consumer = ONSFactory.createConsumer(properties);
       // Subscribe to specific tags. Use "||" as a separator.
       consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() {
           public Action consume(Message message, ConsumeContext context) {
               System.out.println("Receive: " + message);
               return Action.CommitMessage;
           }
       });

       // Subscribe to another topic with all tags.
       // To unsubscribe, remove the subscription code and restart the consumer.
       consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() {
           public Action consume(Message message, ConsumeContext context) {
               System.out.println("Receive: " + message);
               return Action.CommitMessage;
           }
       });

       consumer.start();
       System.out.println("Consumer Started");
   }
}

Consumidor push (consumo por lotes)

Importante

El consumo por lotes requiere la versión 1.8.7.3 o posterior del SDK de cliente TCP para Java. Para obtener detalles sobre las versiones, consulte Release notes.

Dos propiedades controlan el comportamiento de los lotes:

Propiedad	Descripción	Valores válidos	Predeterminado
`ConsumeMessageBatchMaxSize`	Número máximo de mensajes por lote. El SDK invoca la devolución de llamada cuando la cantidad almacenada en caché alcanza este valor.	De 1 a 1024	32
`BatchConsumeMaxAwaitDurationInSeconds`	Tiempo máximo de espera en segundos antes de entregar un lote parcial. El SDK invoca la devolución de llamada al transcurrir esta duración, sin importar el tamaño del lote.	De 0 a 450	0

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import java.util.List;
import java.util.Properties;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.tcp.example.MqConfig;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        consumerProperties.setProperty(PropertyKeyConst.GROUP_ID, MqConfig.GROUP_ID);
        // Read AccessKey credentials from environment variables.
        consumerProperties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        consumerProperties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint from the Instance Details page in the console.
        consumerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MqConfig.NAMESRV_ADDR);

        // Set the maximum batch size to 128 messages.
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // Set the maximum wait time between batches to 10 seconds.
        consumerProperties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, String.valueOf(10));

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // Process messages in batches.
                return Action.CommitMessage;
            }
        });
        // Start the batch consumer.
        batchConsumer.start();
        System.out.println("Consumer start success.");

        // Keep the process alive to continue consuming.
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Consumidor pull

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PullConsumer;
import com.aliyun.openservices.ons.api.TopicPartition;
import java.util.List;
import java.util.Properties;
import java.util.Set;

public class PullConsumerClient {
    public static void main(String[] args){
        Properties properties = new Properties();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        properties.setProperty(PropertyKeyConst.GROUP_ID, "<your-group-id>");
        // Read AccessKey credentials from environment variables.
        properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint from the Instance Details page in the console.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
        PullConsumer consumer = ONSFactory.createPullConsumer(properties);
        // Start the pull consumer.
        consumer.start();
        // Get all partitions for the topic.
        Set<TopicPartition> topicPartitions = consumer.topicPartitions("topic-xxx");
        // Assign partitions to this consumer.
        consumer.assign(topicPartitions);

        while (true) {
            // Poll for messages with a 3-second timeout.
            List<Message> messages = consumer.poll(3000);
            System.out.printf("Received message: %s %n", messages);
        }
    }
}

Para obtener más información sobre particiones y desplazamientos, consulte Terms.

Reintento de mensajes

Si el consumo de mensajes falla o se agota el tiempo de espera, ApsaraMQ for RocketMQ vuelve a entregar el mensaje automáticamente. Para obtener más información, consulte Message retry.