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

Este tema describe cómo suscribirse a mensajes mediante el SDK de cliente TCP para .NET proporcionado por ApsaraMQ for RocketMQ.

Nota

Mantenga la coherencia en las suscripciones de todas las instancias de consumidor identificadas con el mismo ID de grupo. Para obtener más información, consulte Subscription consistency.

ApsaraMQ for RocketMQ admite los siguientes modos de suscripción:

Suscripción en clúster

Todos los consumidores identificados con el mismo ID de grupo consumen una cantidad equitativa de mensajes. Por ejemplo, si un tema contiene nueve mensajes y un grupo de consumidores tiene tres consumidores, en el modo de consumo en clúster cada consumidor procesa tres mensajes.
```
    // Configure clustering subscription, which is the default mode.
    factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.CLUSTERING);              
```
Suscripción en difusión

Cada consumidor identificado con el mismo ID de grupo consume todos los mensajes una vez. Por ejemplo, si un tema contiene nueve mensajes y un grupo de consumidores tiene tres consumidores, en el modo de consumo en difusión cada consumidor procesa los nueve mensajes.
```
    // Configure broadcasting subscription.
    factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.BROADCASTING);                
```

Código de ejemplo:

using System;
using System.Threading;
using System.Text;
using ons;

// The callback function that is executed when a message is pulled from the ApsaraMQ for RocketMQ broker. 
public class MyMsgListener : MessageListener
{
    public MyMsgListener()
    {
    }

    ~MyMsgListener()
    {
    }

    public override ons.Action consume(Message value, ConsumeContext context)
    {
        Byte[] text = Encoding.Default.GetBytes(value.getBody());
        Console.WriteLine(Encoding.UTF8.GetString(text));
        return ons.Action.CommitMessage;
    }
}

public class ConsumerExampleForEx
{
    public ConsumerExampleForEx()
    {
    }

    static void Main(string[] args) {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
        // The AccessKey ID that is used for authentication. 
        factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
	      // The AccessKey secret that is used for authentication. 
        factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console. 
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "GID_example");
        // The topic that you created in the ApsaraMQ for RocketMQ console. 
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "T_example_topic_name");
        // The TCP endpoint. You can obtain the endpoint in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console. 
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "NameSrv_Addr");
        // The log path. 
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");
        // Clustering consumption. 
        // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.CLUSTERING);
        // Broadcasting consumption. 
        // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.BROADCASTING);

        // Create the consumer instance. 
        PushConsumer consumer = ONSFactory.getInstance().createPushConsumer(factoryInfo);

        // Subscribe to the topic. 
        consumer.subscribe(factoryInfo.getPublishTopics(), "*", new MyMsgListener());

        // Start the consumer instance. 
        consumer.start();

        // This setting is used only in this demo. In actual production environments, you cannot exit the process. 
        Thread.Sleep(300000);

        // Before you exit the process, terminate the consumer instance. 
        consumer.shutdown();
    }
}