ApsaraMQ for RocketMQ:Send and subscribe to normal messages with TCP client SDKs

Java

The Java example sends a message asynchronously and returns the result through a callback.

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ProducerTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // AccessKey ID for authentication.
        properties.put(PropertyKeyConst.AccessKey, "<YOUR_ACCESS_KEY>");
        // AccessKey secret for authentication.
        properties.put(PropertyKeyConst.SecretKey, "<YOUR_SECRET_KEY>");
        // Timeout for sending a message, in milliseconds.
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
        // TCP endpoint of your instance.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<YOUR_TCP_ENDPOINT>");

        Producer producer = ONSFactory.createProducer(properties);
        // Call start() once before sending messages.
        producer.start();

        Message msg = new Message(
                // Topic for the message.
                "<YOUR_TOPIC>",
                // Message tag for filtering on the broker.
                "TagA",
                // Message body in binary format.
                // Producer and consumer must agree on serialization and deserialization.
                "Hello MQ".getBytes());

        // Message key: a business-specific identifier, ideally globally unique.
        // Useful for querying or resending messages in the console if delivery fails.
        // Messages can be sent without a key.
        msg.setKey("ORDERID_100");

        // Send the message asynchronously. The result returns through the callback.
        producer.sendAsync(msg, new SendCallback() {
            @Override
            public void onSuccess(final SendResult sendResult) {
                System.out.println("send message success. topic=" +
                    sendResult.getTopic() + ", msgId=" + sendResult.getMessageId());
            }

            @Override
            public void onException(OnExceptionContext context) {
                // Handle the failure: retry or persist the message.
                System.out.println("send message failed. topic=" +
                    context.getTopic() + ", msgId=" + context.getMessageId());
            }
        });

        // The msgId is available before the callback returns.
        System.out.println("send message async. topic=" +
            msg.getTopic() + ", msgId=" + msg.getMsgID());

        // Shut down the producer before exiting. (Optional for Java)
        producer.shutdown();
    }
}

Expected output:

send message async. topic=<YOUR_TOPIC>, msgId=...
send message success. topic=<YOUR_TOPIC>, msgId=...

.NET

The .NET example sends messages synchronously in a loop.

using System;
using ons;

public class ProducerExampleForEx
{
    public ProducerExampleForEx()
    {
    }

    static void Main(string[] args) {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // AccessKey ID for authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "<YOUR_ACCESS_KEY>");
        // AccessKey secret for authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "<YOUR_SECRET_KEY>");
        // Group ID created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ProducerId, "<YOUR_GROUP_ID>");
        // Topic created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<YOUR_TOPIC>");
        // TCP endpoint of your instance.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<YOUR_TCP_ENDPOINT>");
        // Log path.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

        // Create the producer instance.
        // A producer instance is thread-safe and can send messages to different topics.
        // Typically, one producer instance per thread is sufficient.
        Producer producer = ONSFactory.getInstance().createProducer(factoryInfo);

        // Start the producer.
        producer.start();

        // Build and send the message.
        Message msg = new Message(factoryInfo.getPublishTopics(), "tagA", "Example message body");
        msg.setKey(Guid.NewGuid().ToString());
        for (int i = 0; i < 32; i++) {
            try
            {
                SendResultONS sendResult = producer.send(msg);
                Console.WriteLine("send success {0}", sendResult.getMessageId());
            }
            catch (Exception ex)
            {
                Console.WriteLine("send failure{0}", ex.ToString());
            }
        }

        // Shut down the producer before exiting the thread.
        producer.shutdown();
    }
}

Expected output:

send success <message-id>
send success <message-id>
...

C/C++

The C/C++ example sends a message synchronously with exception handling.

#include "ONSFactory.h"
#include "ONSClientException.h"

using namespace ons;

int main()
{
    // Configure the producer.
    ONSFactoryProperty factoryInfo;
    // Group ID created in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "<YOUR_GROUP_ID>");
    // TCP endpoint of your instance.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<YOUR_TCP_ENDPOINT>");
    // Topic created in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics, "<YOUR_TOPIC>");
    // Message content.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "Hello MQ");
    // AccessKey ID for authentication.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<YOUR_ACCESS_KEY>");
    // AccessKey secret for authentication.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<YOUR_SECRET_KEY>");

    // Create the producer.
    Producer *pProducer = ONSFactory::getInstance()->createProducer(factoryInfo);

    // Call start() once before sending messages.
    pProducer->start();

    Message msg(
            // Topic for the message.
            factoryInfo.getPublishTopics(),
            // Message tag for filtering on the broker.
            "TagA",
            // Message body. Cannot be empty.
            // Producer and consumer must agree on serialization and deserialization.
            factoryInfo.getMessageContent()
    );

    // Message key: a business-specific identifier, ideally globally unique.
    // Useful for querying or resending messages in the console if delivery fails.
    // Messages can be sent without a key.
    msg.setKey("ORDERID_100");

    // Send the message. If no exception is thrown, the send succeeded.
    try
    {
        SendResultONS sendResult = pProducer->send(msg);
    }
    catch(ONSClientException & e)
    {
        // Handle the exception: log the error or retry.
    }

    // Shut down the producer before exiting.
    // Failing to shut down may cause memory leaks.
    pProducer->shutdown();

    return 0;
}

Java

The Java consumer subscribes to messages from one or more topics. Filter by specific tags with the "TagA||TagB" syntax, or use "*" to receive all tags.

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();
       // Group ID created in the ApsaraMQ for RocketMQ console.
       properties.put(PropertyKeyConst.GROUP_ID, "<YOUR_GROUP_ID>");
       // AccessKey ID for authentication.
       properties.put(PropertyKeyConst.AccessKey, "<YOUR_ACCESS_KEY>");
       // AccessKey secret for authentication.
       properties.put(PropertyKeyConst.SecretKey, "<YOUR_SECRET_KEY>");
       // TCP endpoint of your instance.
       properties.put(PropertyKeyConst.NAMESRV_ADDR, "<YOUR_TCP_ENDPOINT>");

       // Consumption mode (optional):
       // Clustering (default): each message is delivered to one consumer in the group.
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
       // Broadcasting: each message is delivered to every consumer in the group.
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

       Consumer consumer = ONSFactory.createConsumer(properties);

       // Subscribe to specific tags on a topic.
       consumer.subscribe("<YOUR_TOPIC>", "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");
   }
}

Expected output:

Consumer Started
Receive: <message-details>

.NET

The .NET consumer uses a push consumer with a custom message listener.

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

// Message listener invoked when a 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) {
        ONSFactoryProperty factoryInfo = new ONSFactoryProperty();
        // AccessKey ID for authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.AccessKey, "<YOUR_ACCESS_KEY>");
        // AccessKey secret for authentication.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.SecretKey, "<YOUR_SECRET_KEY>");
        // Group ID created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.ConsumerId, "<YOUR_GROUP_ID>");
        // Topic created in the ApsaraMQ for RocketMQ console.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.PublishTopics, "<YOUR_TOPIC>");
        // TCP endpoint of your instance.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.NAMESRV_ADDR, "<YOUR_TCP_ENDPOINT>");
        // Log path.
        factoryInfo.setFactoryProperty(ONSFactoryProperty.LogPath, "C://log");

        // Consumption mode (optional):
        // Clustering (default): each message is delivered to one consumer in the group.
        // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.CLUSTERING);
        // Broadcasting: each message is delivered to every consumer in the group.
        // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty.BROADCASTING);

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

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

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

        // Keep the process alive to receive messages. In production, do not exit.
        Thread.Sleep(300000);

        // Shut down the consumer before exiting.
        consumer.shutdown();
    }
}

C/C++

The C/C++ consumer uses a push consumer with a message listener callback.

#include "ONSFactory.h"
using namespace ons;

// Message listener that processes each received message.
class MyMsgListener : public MessageListener
{
    public:
        MyMsgListener()
        {
        }

        virtual ~MyMsgListener()
        {
        }

        virtual Action consume(Message &message, ConsumeContext &context)
        {
            // Add your message processing logic here.
            return CommitMessage;
        }
};


int main(int argc, char* argv[])
{
    // Configure the push consumer.
    ONSFactoryProperty factoryInfo;
    // Group ID created in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::ConsumerId, "<YOUR_GROUP_ID>");
    // TCP endpoint of your instance.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "<YOUR_TCP_ENDPOINT>");
    // Topic created in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics, "<YOUR_TOPIC>");
    // AccessKey ID for authentication.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "<YOUR_ACCESS_KEY>");
    // AccessKey secret for authentication.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "<YOUR_SECRET_KEY>");

    // Consumption mode (optional):
    // Clustering (default): each message is delivered to one consumer in the group.
    // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty::CLUSTERING);
    // Broadcasting: each message is delivered to every consumer in the group.
    // factoryInfo.setFactoryProperty(ONSFactoryProperty:: MessageModel, ONSFactoryProperty::BROADCASTING);

    // Create the push consumer.
    PushConsumer* pushConsumer = ONSFactory::getInstance()->createPushConsumer(factoryInfo);

    // Subscribe to the topic and register the message listener.
    MyMsgListener msglistener;
    pushConsumer->subscribe(factoryInfo.getPublishTopics(), "*", &msglistener);

    // Start the push consumer.
    pushConsumer->start();

    // Keep the process alive to receive messages.
    // Call shutdown() only when the consumer is idle (not processing messages).
    // Failing to shut down may cause memory leaks.
    pushConsumer->shutdown();
    return 0;
}