Sample code
Updated at:
ApsaraMQ for RocketMQ 5.x instances support clients that use C++ SDK 1.x. This topic provides sample code that shows how to use C++ SDK 1.x to connect to a 5.x instance to send and receive messages.
Important
We recommend that you use the latest RocketMQ 5.x SDKs. These SDKs are fully compatible with ApsaraMQ for RocketMQ 5.x brokers and provide more functions and enhanced features. For more information, see Version guide.
Alibaba Cloud only maintains RocketMQ 3.x, 4.x, and TCP client SDKs. We recommend that you use them only for existing business.
Send and receive normal messages
Send normal messages
#include "ONSFactory.h"
#include "ONSClientException.h"
using namespace ons;
int main()
{
// Create a producer and configure the required information for sending messages.
ONSFactoryProperty factoryInfo;
// Set this to the group ID that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
// Set this to the endpoint that you obtained from the ApsaraMQ for RocketMQ console. The format is similar to "rmq-cn-XXXX.rmq.aliyuncs.com:8080".
// Note: Enter the domain name and port provided in the console. Do not add the http:// or https:// prefix. Do not use a resolved IP address.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "ACCESS POINT");
// The topic that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
// The message content.
factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
/**
* If you use a public endpoint, you must set the AccessKey and SecretKey. Enter the instance username and password. You can obtain the instance username and password from the Intelligent Authentication tab on the Access Control page in the console.
* Note: Do not use the AccessKey ID and AccessKey secret of your Alibaba Cloud account.
* If you access the instance from an Alibaba Cloud ECS instance over an internal network, you do not need to configure these parameters. The server automatically obtains the information based on the VPC.
* If you use a Serverless instance, you must set the instance username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
*/
// Obtain the instance username and password from the Intelligent Authentication tab of the Access Control page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "INSTANCE USER NAME");
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "INSTANCE PASSWORD" );
// Note: When you access an ApsaraMQ for RocketMQ 5.x instance, do not set the InstanceID property. Otherwise, the access fails.
// Create a producer.
Producer *pProducer = ONSFactory::getInstance()->createProducer(factoryInfo);
// Before sending messages, call the start method to start the producer. This method needs to be called only once.
pProducer->start();
Message msg(
// Message topic.
factoryInfo.getPublishTopics(),
// Message tag. This is similar to a label in Gmail. It is used to sub-categorize messages, which helps consumers specify filter conditions on the ApsaraMQ for RocketMQ server.
"TagA",
// Message body. This cannot be empty. ApsaraMQ for RocketMQ does not interfere with the message body. The producer and consumer must agree on a consistent serialization and deserialization method.
factoryInfo.getMessageContent()
);
// Set the message key. The key is a business-specific attribute of a message. Make the key globally unique if possible.
// This helps you query for the message and resend it from the ApsaraMQ for RocketMQ console if it is not received properly.
// Note: Not setting the key does not affect normal message sending and receiving.
msg.setKey("ORDERID_100");
// Send the message. If no exception is thrown, the message is sent successfully.
try
{
SendResultONS sendResult = pProducer->send(msg);
}
catch(ONSClientException & e)
{
// Handle the exception details as needed.
}
// Before the application exits, destroy the producer object. Otherwise, issues such as memory leaks may occur.
pProducer->shutdown();
return 0;
}Subscribe to normal messages
#include "ONSFactory.h"
#include <iostream>
#include <thread>
#include <mutex>
using namespace ons;
std::mutex console_mtx;
class ExampleMessageListener : public MessageListener {
public:
Action consume(Message& message, ConsumeContext& context) {
// This is the message processing procedure. Return CommitMessage to confirm that the message was processed successfully.
// If a consumption exception occurs or you want to reconsume the message, return ReconsumeLater. The message will be redelivered after a period of time.
std::lock_guard<std::mutex> lk(console_mtx);
std::cout << "Received a message. Topic: " << message.getTopic() << ", MsgId: "
<< message.getMsgID() << std::endl;
return CommitMessage;
}
};
int main(int argc, char* argv[]) {
std::cout << "=======Before consuming messages=======" << std::endl;
ONSFactoryProperty factoryInfo;
// Set this to the group ID that you created in the ApsaraMQ for RocketMQ console. Since the instantiated version, ProducerId and ConsumerId have been unified. This setting is for forward compatibility.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ConsumerId, "GID_XXX");
/**
* If you use a public endpoint, you must set the AccessKey and SecretKey. Enter the instance username and password. You can obtain the instance username and password from the Intelligent Authentication tab on the Access Control page in the console.
* Note: Do not use the AccessKey ID and AccessKey secret of your Alibaba Cloud account.
* If you access the instance from an Alibaba Cloud ECS instance over an internal network, you do not need to configure these parameters. The server automatically obtains the information based on the VPC.
* If you use a Serverless instance, you must set the instance username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
*/
// Obtain the instance username and password from the Intelligent Authentication tab of the Access Control page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "INSTANCE USER NAME");
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "INSTANCE PASSWORD" );
// Set this to the endpoint that you obtained from the ApsaraMQ for RocketMQ console. The format is similar to "rmq-cn-XXXX.rmq.aliyuncs.com:8080".
// Note: Enter the domain name and port provided in the console. Do not add the http:// or https:// prefix. Do not use a resolved IP address.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "ACCESS POINT");
// Note: When you access an ApsaraMQ for RocketMQ 5.x instance, do not set the InstanceID property. Otherwise, the access fails.
PushConsumer *consumer = ONSFactory::getInstance()->createPushConsumer(factoryInfo);
// Set this to the topic that you created in the ApsaraMQ for RocketMQ console.
const char* topic_1 = "topic-1";
// Subscribe to all messages in topic-1 that have the Tag message attribute set to tag-1.
const char* tag_1 = "tag-1";
const char* topic_2 = "topic-2";
// Subscribe to all messages in topic-2.
const char* tag_2 = "*";
// Register a custom listener function to process received messages and return the processing result.
ExampleMessageListener * message_listener = new ExampleMessageListener();
consumer->subscribe(topic_1, tag_1, message_listener);
consumer->subscribe(topic_2, tag_2, message_listener);
// After the preparation is complete, you must call the start function to start the consumer.
consumer->start();
// Keep the thread running. Do not perform the shutdown operation.
std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));
consumer->shutdown();
delete message_listener;
std::cout << "=======After consuming messages======" << std::endl;
return 0;
}Send and receive ordered messages
Send ordered messages
#include "ONSFactory.h"
#include "ONSClientException.h"
#include <iostream>
using namespace ons;
int main()
{
// Required parameters for creating and running the producer.
ONSFactoryProperty factoryInfo;
// Set this to the group ID that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
// Set this to the endpoint that you obtained from the ApsaraMQ for RocketMQ console. The format is similar to "rmq-cn-XXXX.rmq.aliyuncs.com:8080".
// Note: Enter the domain name and port provided in the console. Do not add the http:// or https:// prefix. Do not use a resolved IP address.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "ACCESS POINT");
// The topic that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
// The message content.
factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
/**
* If you use a public endpoint, you must set the AccessKey and SecretKey. Enter the instance username and password. You can obtain the instance username and password from the Intelligent Authentication tab on the Access Control page in the console.
* Note: Do not use the AccessKey ID and AccessKey secret of your Alibaba Cloud account.
* If you access the instance from an Alibaba Cloud ECS instance over an internal network, you do not need to configure these parameters. The server automatically obtains the information based on the VPC.
* If you use a Serverless instance, you must set the instance username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
*/
// Obtain the instance username and password from the Intelligent Authentication tab of the Access Control page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "INSTANCE USER NAME");
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "INSTANCE PASSWORD" );
// Note: When you access an ApsaraMQ for RocketMQ 5.x instance, do not set the InstanceID property. Otherwise, the access fails.
// Create a producer.
OrderProducer *pProducer = ONSFactory::getInstance()->createOrderProducer(factoryInfo);
// Before sending messages, call the start method to start the producer. This method needs to be called only once.
pProducer->start();
Message msg(
// Message topic.
factoryInfo.getPublishTopics(),
// Message tag. This is similar to a label in Gmail. It is used to sub-categorize messages, which helps consumers specify filter conditions on the ApsaraMQ for RocketMQ server.
"TagA",
// Message body. This is any data in binary format. ApsaraMQ for RocketMQ does not interfere with the message body. The producer and consumer must agree on a consistent serialization and deserialization method.
factoryInfo.getMessageContent()
);
// Set the message key. The key is a business-specific attribute of a message. Make the key globally unique if possible.
// This helps you query for the message and resend it from the ApsaraMQ for RocketMQ console if it is not received properly.
// Note: Not setting the key does not affect normal message sending and receiving.
msg.setKey("ORDERID_100");
// The key field that distinguishes different partitions in partitionally ordered messages.
// For globally ordered messages, this field can be set to any non-empty string.
std::string shardingKey = "abc";
// Messages with the same sharding key are sent in order.
try
{
// Send the message. If no exception is thrown, the message is sent successfully.
SendResultONS sendResult = pProducer->send(msg, shardingKey);
std::cout << "send success" << std::endl;
}
catch(ONSClientException & e)
{
// Add exception handling.
}
// Before the application exits, destroy the producer object. Otherwise, issues such as memory leaks may occur.
pProducer->shutdown();
return 0;
} Subscribe to ordered messages
#include "ONSFactory.h"
using namespace std;
using namespace ons;
// Create an instance to consume messages.
// After the pushConsumer pulls a message, it automatically calls the consumeMessage function of this instance.
class ONSCLIENT_API MyMsgListener : public MessageOrderListener
{
public:
MyMsgListener()
{
}
virtual ~MyMsgListener()
{
}
virtual OrderAction consume(Message &message, ConsumeOrderContext &context)
{
// Consume the message as needed.
return Success; //CONSUME_SUCCESS;
}
};
int main(int argc, char* argv[])
{
// Required parameters for creating and running the OrderConsumer.
ONSFactoryProperty factoryInfo;
// Set this to the group ID that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ConsumerId, "XXX");
// Set this to the topic that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
/**
* If you use a public endpoint, you must set the AccessKey and SecretKey. Enter the instance username and password. You can obtain the instance username and password from the Intelligent Authentication tab on the Access Control page in the console.
* Note: Do not use the AccessKey ID and AccessKey secret of your Alibaba Cloud account.
* If you access the instance from an Alibaba Cloud ECS instance over an internal network, you do not need to configure these parameters. The server automatically obtains the information based on the VPC.
* If you use a Serverless instance, you must set the instance username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
*/
// Obtain the instance username and password from the Intelligent Authentication tab of the Access Control page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "INSTANCE USER NAME");
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "INSTANCE PASSWORD" );
// Set this to the endpoint that you obtained from the ApsaraMQ for RocketMQ console. The format is similar to "rmq-cn-XXXX.rmq.aliyuncs.com:8080".
// Note: Enter the domain name and port provided in the console. Do not add the http:// or https:// prefix. Do not use a resolved IP address.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "ACCESS POINT");
// Note: When you access an ApsaraMQ for RocketMQ 5.x instance, do not set the InstanceID property. Otherwise, the access fails.
// Create an orderConsumer.
OrderConsumer* orderConsumer = ONSFactory::getInstance()->createOrderConsumer(factoryInfo);
MyMsgListener msglistener;
// Specify the message topic and message tag for the orderConsumer to subscribe to.
orderConsumer->subscribe(factoryInfo.getPublishTopics(), "*",&msglistener );
// Register the message listener handler instance. After the orderConsumer pulls a message, it calls the consumeMessage function of this class.
// Start the orderConsumer.
orderConsumer->start();
for(volatile int i = 0; i < 10; ++i) {
// wait
}
// Destroy the orderConsumer. Before the application exits, you must destroy the consumer object. Otherwise, issues such as memory leaks may occur.
orderConsumer->shutdown();
return 0;
} Send and receive scheduled and delayed messages
Send scheduled and delayed messages
#include "ONSFactory.h"
#include "ONSClientException.h"
#include <windows.h>
using namespace ons;
int main()
{
// Create a producer and configure the required information for sending messages.
ONSFactoryProperty factoryInfo;
// Set this to the group ID that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
// Set this to the endpoint that you obtained from the ApsaraMQ for RocketMQ console. The format is similar to "rmq-cn-XXXX.rmq.aliyuncs.com:8080".
// Note: Enter the domain name and port provided in the console. Do not add the http:// or https:// prefix. Do not use a resolved IP address.
factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "ACCESS POINT");
// The topic that you created in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
// The message content.
factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
/**
* If you use a public endpoint, you must set the AccessKey and SecretKey. Enter the instance username and password. You can obtain the instance username and password from the Intelligent Authentication tab on the Access Control page in the console.
* Note: Do not use the AccessKey ID and AccessKey secret of your Alibaba Cloud account.
* If you access the instance from an Alibaba Cloud ECS instance over an internal network, you do not need to configure these parameters. The server automatically obtains the information based on the VPC.
* If you use a Serverless instance, you must set the instance username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
*/
// Obtain the instance username and password from the Intelligent Authentication tab of the Access Control page in the ApsaraMQ for RocketMQ console.
factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "INSTANCE USER NAME");
factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "INSTANCE PASSWORD" );
// Note: When you access an ApsaraMQ for RocketMQ 5.x instance, do not set the InstanceID property. Otherwise, the access fails.
// Create a producer.
Producer *pProducer = ONSFactory::getInstance()->createProducer(factoryInfo);
// Before sending messages, call the start method to start the producer. This method needs to be called only once.
pProducer->start();
Message msg(
// Message topic.
factoryInfo.getPublishTopics(),
// Message tag. This is similar to a label in Gmail. It is used to sub-categorize messages, which helps consumers specify filter conditions on the ApsaraMQ for RocketMQ server.
"TagA",
// Message body. This cannot be empty. ApsaraMQ for RocketMQ does not interfere with the message body. The producer and consumer must agree on a consistent serialization and deserialization method.
factoryInfo.getMessageContent()
);
// Set the message key. The key is a business-specific attribute of a message. Make the key globally unique if possible.
// This helps you query for the message and resend it from the ApsaraMQ for RocketMQ console if it is not received properly.
// Note: Not setting the key does not affect normal message sending and receiving.
msg.setKey("ORDERID_100");
// The unit of deliver time is ms. Specify a point in time after which the message can be consumed. This example indicates that the message can be consumed after 3 seconds.
long deliverTime = GetTickCount64() + 3000;
msg.setStartDeliverTime(deliverTime);
// Send the message. If no exception is thrown, the message is sent successfully.
try
{
SendResultONS sendResult = pProducer->send(msg);
}
catch(ONSClientException & e)
{
// Handle the exception details as needed.
}
// Before the application exits, destroy the producer object. Otherwise, issues such as memory leaks may occur.
pProducer->shutdown();
return 0;
}
Subscribe to scheduled and delayed messages
The sample code for subscribing to scheduled and delayed messages is the same as that for subscribing to normal messages. For more information, see Subscribe to normal messages.
Send and receive transactional messages
Send transactional messages
Send a half message and execute the local transaction. The sample code is as follows.
#include "ONSFactory.h" #include "ONSClientException.h" using namespace ons; class MyLocalTransactionExecuter : LocalTransactionExecuter { MyLocalTransactionExecuter() { } ~MyLocalTransactionExecuter() { } virtual TransactionStatus execute(Message &value) { // Message ID. The message body may be the same, but the message ID is different. The current message ID cannot be queried in the ApsaraMQ for RocketMQ console. string msgId = value.getMsgID(); // Perform a crc32 check on the message body content. You can also use other algorithms such as MD5. // The message ID and crc32id are mainly used to prevent message duplication. // If the business is idempotent, you can ignore this. Otherwise, you need to use msgId or crc32Id to ensure idempotence. // If messages must not be duplicated, use crc32 or MD5 on the message body to prevent duplicate messages. TransactionStatus transactionStatus = Unknow; try { boolean isCommit = local_transaction_execution_result; if (isCommit) { // The local transaction is successful. Commit the message. transactionStatus = CommitTransaction; } else { // The local transaction failed. Roll back the message. transactionStatus = RollbackTransaction; } } catch (...) { //exception handle } return transactionStatus; } } int main(int argc, char* argv[]) { //Create a producer and configure the required information for sending messages. ONSFactoryProperty factoryInfo; // Set this to the group ID that you created in the ApsaraMQ for RocketMQ console. factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX"); // Set this to the endpoint that you obtained from the ApsaraMQ for RocketMQ console. The format is similar to "rmq-cn-XXXX.rmq.aliyuncs.com:8080". // Note: Enter the domain name and port provided in the console. Do not add the http:// or https:// prefix. Do not use a resolved IP address. factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "ACCESS POINT"); // The topic that you created in the ApsaraMQ for RocketMQ console. factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" ); // The message content. factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX"); /** * If you use a public endpoint, you must set the AccessKey and SecretKey. Enter the instance username and password. You can obtain the instance username and password from the Intelligent Authentication tab on the Access Control page in the console. * Note: Do not use the AccessKey ID and AccessKey secret of your Alibaba Cloud account. * If you access the instance from an Alibaba Cloud ECS instance over an internal network, you do not need to configure these parameters. The server automatically obtains the information based on the VPC. * If you use a Serverless instance, you must set the instance username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access. */ // Obtain the instance username and password from the Intelligent Authentication tab of the Access Control page in the ApsaraMQ for RocketMQ console. factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, "INSTANCE USER NAME"); factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, "INSTANCE PASSWORD" ); // Note: When you access an ApsaraMQ for RocketMQ 5.x instance, do not set the InstanceID property. Otherwise, the access fails. // Create a producer. ApsaraMQ for RocketMQ is not responsible for releasing pChecker. You must release the resource yourself. MyLocalTransactionChecker *pChecker = new MyLocalTransactionChecker(); g_producer = ONSFactory::getInstance()->createTransactionProducer(factoryInfo,pChecker); // Before sending messages, call the start method to start the producer. This method needs to be called only once. pProducer->start(); Message msg( // Message topic. factoryInfo.getPublishTopics(), // Message tag. This is similar to a label in Gmail. It is used to sub-categorize messages, which helps consumers specify filter conditions on the ApsaraMQ for RocketMQ server. "TagA", // Message body. This cannot be empty. ApsaraMQ for RocketMQ does not interfere with the message body. The producer and consumer must agree on a consistent serialization and deserialization method. factoryInfo.getMessageContent() ); // Set the message key. The key is a business-specific attribute of a message. Make the key globally unique if possible. // This helps you query for the message and resend it from the ApsaraMQ for RocketMQ console if it is not received properly. // Note: Not setting the key does not affect normal message sending and receiving. msg.setKey("ORDERID_100"); // Send the message. If no exception is thrown, the message is sent successfully. try { //ApsaraMQ for RocketMQ is not responsible for releasing pExecuter. You must release the resource yourself. MyLocalTransactionExecuter pExecuter = new MyLocalTransactionExecuter(); SendResultONS sendResult = pProducer->send(msg,pExecuter); } catch(ONSClientException & e) { // Handle the exception details as needed. } // Before the application exits, destroy the producer object. Otherwise, issues such as memory leaks may occur. pProducer->shutdown(); return 0; }Commit the status of a transactional message. The sample code is as follows.
class MyLocalTransactionChecker : LocalTransactionChecker { MyLocalTransactionChecker() { } ~MyLocalTransactionChecker() { } virtual TransactionStatus check(Message &value) { // Message ID. The message body may be the same, but the message ID is different. The current message ID cannot be queried in the ApsaraMQ for RocketMQ console. string msgId = value.getMsgID(); // Perform a crc32 check on the message body content. You can also use other algorithms such as MD5. // The message ID and crc32id are mainly used to prevent message duplication. // If the business is idempotent, you can ignore this. Otherwise, you need to use msgId or crc32Id to ensure idempotence. // If messages must not be duplicated, use crc32 or MD5 on the message body to prevent duplicate messages. TransactionStatus transactionStatus = Unknow; try { boolean isCommit = local_transaction_execution_result; if (isCommit) { // The local transaction is successful. Commit the message. transactionStatus = CommitTransaction; } else { // The local transaction failed. Roll back the message. transactionStatus = RollbackTransaction; } } catch(...) { //exception error } return transactionStatus; } }
Subscribe to transactional messages
The sample code for subscribing to transactional messages is the same as that for subscribing to normal messages. For more information, see Subscribe to normal messages.
Is this page helpful?