消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 4.xC++ SDK顺序消息
复制全文
下载 pdf顺序消息
文档反馈
问问助手消息队列 RocketMQ版提供顺序消息（FIFO 消息）供您使用。在顺序消息模型中，您需要严格按照顺序来发布和消费消息。本文提供使用 C++ SDK 收发顺序消息的示例代码供您参考。
背景信息
顺序消息分为两类，全局顺序消息和分区顺序消息。区别仅为队列数量不同，代码没有区别。
- 全局顺序消息：
- 对于指定的一个 Topic，所有消息的生产和消费需要遵循一定的顺序，消息的消费顺序必须和生产顺序一致，即需要严格的先入先出 FIFO（First In First Out）的顺序进行发布和消费。
- 分区顺序消息：
- 对于指定的一个 Topic，其中每一个分区的消息生产与消费是有序的，同一个队列内的消息按照严格的 FIFO 顺序进行发布和订阅。消息投递到哪一个分区由消息的 Sharding Key 来进行区分。在 SDK 中可以通过指定 Sharding Key 和 MessageQueueSelector 回调函数来控制消息投递到哪个分区。
前提条件
- 已完成准备工作。
- 已阅读参数说明，了解常用参数的配置方式与填写格式。
发送顺序消息
发送顺序消息的示例代码如下。使用 g++ 命令进行编译。
g++ -o order_producer order_producer.cpp -lrocketmq -lpthread -lz -ldl -lrt
如果页面提示 undefined reference 相关的编译错误，请先确定是否已安装动态库，如果确定已安装，可以尝试在 g++ 命令添加 -D_GLIBCXX_USE_CXX11_ABI=0 参数重新尝试编译。
include<iostream>
include<chrono>
include<thread>
include"rocketmq/DefaultMQProducer.h"
using namespace std;
using namespace rocketmq;
class DefineSelectMessageQueue : public MessageQueueSelector
{
public:
MQMessageQueue select(const std::vector<MQMessageQueue> &mqs, const MQMessage &msg, void *arg)
{
// 实现自定义分区逻辑，根据业务传入arg参数即分区键，计算路由到哪个队列，这里以arg为int型参数为例。
int orderId = *static_cast<int *>(arg);
int index = orderId % mqs.size();
return mqs[index];
}
};
int main()
{
// 生产者名称无需申请
DefaultMQProducer producer("producer_group_name");
// 火山引擎的接入点
producer.setNamesrvAddr("accesspoint");
// AK、SK替换为RocketMQ实例的AccessKey和AccessKey Secret
// 用户渠道，可以标明和用户相关即可，无需申请。
producer.setSessionCredentials("ak", "sk", "volc");
// 请确保参数设置完成之后启动Producer。
producer.start();
DefineSelectMessageQueue *queueSelector = new DefineSelectMessageQueue();
int count = 64;
for (int i = 0; i < count; ++i)
{
// 配置消息写入的Topic名称。对于实例 ID 格式为 MQ_INST_xxxx 的实例，此处配置的格式为${实例ID%TopicID}，例如 "MQ_INST_****%testTopic"。详细说明请参考https://www.volcengine.com/docs/6410/153010#注意事项。
MQMessage msg("you_topic_name", "TAG", "msg content");
try
{
SendResult sendResult = producer.send(msg, queueSelector, &i, 3, false);
std::cout << "SendResult:" << sendResult.getSendStatus() << ", Message ID: " << sendResult.getMsgId() << std::endl;
this_thread::sleep_for(chrono::seconds(1));
}
catch (MQException e)
{
std::cout << "ErrorCode: " << e.GetError() << " Exception:" << e.what() << std::endl;
}
}
std::cout << "Send " << count << " messages OK, costs" << std::endl;
producer.shutdown();
return 0;
}
订阅顺序消息
订阅顺序消息的示例代码如下。
g++ -o order_consumer order_consumer.cpp -lrocketmq -lpthread -lz -ldl -lrt
如果页面提示 undefined reference 相关的编译错误，请先确定是否已安装动态库，如果确定已安装，可以尝试在 g++ 命令添加 -D_GLIBCXX_USE_CXX11_ABI=0 参数重新尝试编译。
include<iostream>
include<thread>
include"rocketmq/DefaultMQPushConsumer.h"
using namespace rocketmq;
class OrderlyMessageListener : public MessageListenerOrderly
{
public:
ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs)
{
for (auto item = msgs.begin(); item != msgs.end(); item++)
{
std::cout << "Received Message Topic:" << item->getTopic() << ", MsgId:" << item->getMsgId() << std::endl;
}
return CONSUME_SUCCESS;
}
};
int main(int argc, char *argv[])
{
// 您在火山引擎消息队列 RocketMQ控制台上申请的GID。
DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_group");
// 从火山引擎消息队列 RocketMQ控制台的实例详情页面获取。
consumer->setNamesrvAddr("your access point");
// AK、SK替换为RocketMQ实例的AccessKey和AccessKey Secret
// 用户渠道，可以标明和用户相关即可，无需申请。
consumer->setSessionCredentials("ak", "sk", "VOLC");
// 请注册自定义监听函数用来处理接收到的消息，并返回响应的处理结果。
OrderlyMessageListener *messageListener = new OrderlyMessageListener();
consumer->subscribe("topic_name", "tag");
consumer->registerMessageListener(messageListener);
// Start this consumer
// 准备工作完成，必须调用启动函数，才可以正常工作。
// ********************************************
// 1.确保订阅关系的设置在启动之前完成。
// 2.确保相同GID下面的消费者的订阅关系一致。
// *********************************************
consumer->start();
// 保持主线程运行直到进程结束
std::this_thread::sleep_for(std::chrono::seconds(60));
consumer->shutdown();
return 0;
}
最近更新时间：2025.07.09 16:22:42
这个页面对您有帮助吗？
有用
有用
无用
无用