Subscribe to messages

This topic describes how to subscribe to messages by using the .NET SDK of RocketMQ.

Note:

Maintain consistent subscription for all consumer instances with the same group ID. For more information, see Subscription consistency.

Subscription modes

RocketMQ supports the following two message subscription modes:

• Clustering subscription: All the consumers identified by the same group ID equally share messages. For example, a topic contains nine messages and a group contains three consumer instances. In this case, each instance consumes three messages.

   // The configuration of clustering subscription (default mode).
    factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.CLUSTERING);

• Broadcasting subscription: All the consumers identified by the same group ID consume every message once. For example, a topic contains nine messages and a group contains three consumer instances. In this case, each instance consumes nine messages.

   // The configuration of broadcasting subscription.
    factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.BROADCASTING);

Sample code

using System;
using System.Threading;
using System.Text;
using ons;
// The callback function you need to execute when the message is pulled from the 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) {
        // Configure your account according to the following settings. You can obtain these settings in the console.
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // The AccessKeyId you created in the Alibaba Cloud console for identity authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "Your access key");
        // The AccessKeySecret you created in the Alibaba Cloud console for identity authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "Your access secret");
        // The group ID you created in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "GID_example");
        // The topic you created in the console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "T_example_topic_name");
        // Set the TCP endpoint: Go to the **Instances** page in the RocketMQ console, and view the endpoint in the **Endpoint Information** area.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "NameSrv_Addr");
        // Set 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 consumer instances.
        PushConsumer consumer = ONSFactory.getInstance().createPushConsumer(factoryInfo);
        // Subscribe to topics.
        consumer.subscribe(factoryInfo.getPublishTopics(), "*", new MyMsgListener());
        // Start the instance at the client.
        consumer.start();
        //This setting is only used in the demo. In actual production environment, you cannot exit the process.
        Thread.Sleep(300000);
        // Disable the consumer instance when the process is about to exit.
        consumer.shutdown();
    }
}