Subscribe to messages

This topic introduces how to subscribe to messages using .NET SDK.

Note:

• Make sure the subscription relationships of all consumer instances under the same consumer ID are consistent. For detailed information, see Subscription consistency.

Subscription Methods

MQ supports the following two subscription methods:

• Clustering Subscription: All the consumers with the same consumer ID evenly share message consumption. For example, if there are 9 messages under a certain topic, and 3 consumer instances with same consumer ID, then in clustering consumption mode, each instance consumes 3 messages respectively.

   // Clustering subscription settings (it is the default method if not configured)
   factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.CLUSTERING);

• Broadcasting subscription: All the consumers with the same consumer ID consume each message once. For example, if there are 9 messages under a certain topic, and 3 consumer instances with the same consumer ID, then each instance consumes all the 9 messages respectively.

   ## Broadcasting subscription settings
   factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.BROADCASTING);

Sample Code

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Runtime.InteropServices;
using ons;
namespace ons
{
    // When pushConsumer has pulled the message, it will actively invoke the consumer function of the instance.
    public class MyMsgListener : MessageListener
    {    
        public MyMsgListener()
        {
        }
        ~MyMsgListener()
        {
        }
        public override  Action  consume(Message value, ConsumeContext context)
        {
           // The message includes the consumed message, and the message body can be obtained through getBody interface
           Console.WriteLine(value.getBody());
            /*
                   All questions related to Chinese encoding are answered in the document included in the SDK package, please read it carefully
           */
            return ons.Action.CommitMessage;
        }
    }
    class onscsharp
    {
           static void Main(string[] args)
        {
            // Mandatory parameter required for the creation and operation of pushConsumer
            ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
            factoryInfo.setFactoryProperty(factoryInfo.ConsumerId, "XXX");//The consumer ID you have created in the MQ console
            factoryInfo.setFactoryProperty(factoryInfo.PublishTopics, "XXX");//The topic you have created in the MQ console
            factoryInfo.setFactoryProperty(factoryInfo.AccessKey,"xx");//AccessKey, Alibaba Cloud ID verification, which is created on Alibaba Cloud Management Console
            factoryInfo.setFactoryProperty(factoryInfo.SecretKey, "xxxx");//SecretKey, Alibaba Cloud ID verification, which is created on Alibaba Cloud Management Console
            // Clustering subscription (default)
            // factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.CLUSTERING);
            // Broadcasting subscription 
            // factoryInfo.setFactoryProperty(ONSFactoryProperty.MessageModel, ONSFactoryProperty.BROADCASTING);
            // Create a consumer for message consumption
            ONSFactory onsfactory = new ONSFactory();
            PushConsumer pConsumer = onsfactory.getInstance().createPushConsumer(factoryInfo);
            // Register message callback for the consumer, and the callback function will be triggered once there is a message.
            MessageListener msgListener = new MyMsgListener();
            pConsumer.subscribe(factoryInfo.getPublishTopics(), "*",  msgListener);
            // Start the consumer
            pConsumer.start();   
            // When the consumer is started, messages will be pulled asynchronously. When the messages have been pulled, MQ calls back the consumer function of the MyMsgListener instance to pass the message body to the consumer through a parameter.     
            // Here you can continue to code the service related logic. When the consumption is completed, call shutdown function and release the resource. The shutdown function also needs to be called when quitting the application.
            pConsumer.shutdown();
        }
    }
}