RocketMQ 4.x/3.x C++ SDK接入5.x实例收发消息示例代码 - 云消息队列 RocketMQ 版

云消息队列 RocketMQ 版5.x版本实例可兼容4.x/3.x SDK客户端接入，您可以使用4.x/3.x版本的SDK接入5.x实例进行消息收发。本文为您介绍4.x/3.x版本下的C++ SDK消息收发示例代码。

重要

推荐您使用最新的RocketMQ 5.x系列SDK，5.x系列SDK作为主力研发版本，和云消息队列 RocketMQ 版5.x服务端完全兼容，提供了更全面的功能并支持更多增强特性。更多信息，请参见5.x系列SDK。
RocketMQ 4.x/3.x系列SDK和ONS系列SDK后续仅做功能维护，建议仅存量业务使用。

普通消息收发示例

发送普通消息

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

int main() {
    std::cout << "=======Before sending messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。      
    DefaultMQProducer producer("GID_XXX");          
    // 设置为您从阿里云消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
    // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。 
    producer.setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。  
    // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    producer.setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。

    // 请确保参数设置完成之后启动Producer。
    producer.start();
    auto start = std::chrono::system_clock::now();
    int count = 32;
    for (int i = 0; i < count; ++i) {
        // 设置为您在消息队列RocketMQ版控制台上创建的普通消息类型的Topic。
        MQMessage msg("YOURTOPIC","HiTAG","HelloCPPSDK.");
        try {
            SendResult sendResult = producer.send(msg);
            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;
        }
    }
    auto interval = std::chrono::system_clock::now() - start;
    std::cout << "Send " << count << " messages OK, costs "
              << std::chrono::duration_cast<std::chrono::milliseconds>(interval).count() << "ms" << std::endl;

    producer.shutdown();
    std::cout << "=======After sending messages=======" << std::endl;
    return 0;
}

订阅普通消息

#include <iostream>
#include <thread>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;


class ExampleMessageListener : public MessageListenerConcurrently {
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[]) {
    std::cout << "=======Before consuming messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。
    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXX");
    // 设置为您从消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
    // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。
    consumer->setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。
    // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    consumer->setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。
    
    auto start = std::chrono::system_clock::now();

    // 请注册自定义监听函数用来处理接收到的消息，并返回响应的处理结果。
    ExampleMessageListener *messageListener = new ExampleMessageListener();
    consumer->subscribe("YOURTOPIC", "*");
    consumer->registerMessageListener(messageListener);

    // 准备工作完成，必须调用启动函数，才可以正常工作。
    // ********************************************
    // 1.确保订阅关系的设置在启动之前完成。
    // 2.确保相同消费者分组内的消费者的订阅关系一致。
    // *********************************************
    consumer->start();

    // 请保持线程常驻，不要执行shutdown操作。
    std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
}

顺序消息收发示例

发送顺序消息

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

class ExampleSelectMessageQueueByHash : 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[0];
    }
};

int main() {
    std::cout << "=======Before sending messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。
    DefaultMQProducer producer("GID_XXX");
    // 设置为您从消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
    // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。
    producer.setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。
    // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    producer.setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。

    // 请确保参数设置完成之后启动Producer。
    producer.start();
    auto start = std::chrono::system_clock::now();
    int count = 32;
    // 可以设置发送重试的次数，确保发送成功。
    int retryTimes = 1;
    // 参考接口MessageQueueSelector，通过设置的自定义参数arg，计算发送到指定的路由队列中，此处的arg便是分区ID。
    ExampleSelectMessageQueueByHash *pSelector = new ExampleSelectMessageQueueByHash();
    for (int i = 0; i < count; ++i) {
        // 设置为您在消息队列RocketMQ版控制台上创建的顺序消息类型的Topic。
        MQMessage msg("YOUR ORDERLY TOPIC", "HiTAG", "Hello,CPP SDK, Orderly Message.");
        try {
            SendResult sendResult = producer.send(msg, pSelector, &i, 1, false);
            std::cout << "SendResult:" << sendResult.getSendStatus() << ", Message ID: " << sendResult.getMsgId()
                      << "MessageQueue:" << sendResult.getMessageQueue().toString() << std::endl;
            this_thread::sleep_for(chrono::seconds(1));
        } catch (MQException e) {
            std::cout << "ErrorCode: " << e.GetError() << " Exception:" << e.what() << std::endl;
        }
    }
    auto interval = std::chrono::system_clock::now() - start;
    std::cout << "Send " << count << " messages OK, costs "
              << std::chrono::duration_cast<std::chrono::milliseconds>(interval).count() << "ms" << std::endl;

    producer.shutdown();
    std::cout << "=======After sending messages=======" << std::endl;
    return 0;
}

订阅顺序消息


#include <iostream>
#include <thread>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;


class ExampleOrderlyMessageListener : 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[]) {
    std::cout << "=======Before consuming messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。
    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXX");
    // 设置为您从消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
    // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。
    consumer->setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。
    // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    consumer->setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。
 
    auto start = std::chrono::system_clock::now();

    // 请注册自定义侦听函数用来处理接收到的消息，并返回响应的处理结果。
    ExampleOrderlyMessageListener *messageListener = new ExampleOrderlyMessageListener();
    consumer->subscribe("YOUR ORDERLY TOPIC", "*");
    consumer->registerMessageListener(messageListener);

    // 准备工作完成，必须调用启动函数，才可以正常工作。
    // ********************************************
    // 1.确保订阅关系的设置在启动之前完成。
    // 2.确保相同消费者分组内的消费者的订阅关系一致。
    // *********************************************
    consumer->start();

    // 请保持线程常驻，不要执行shutdown操作。
    std::this_thread::sleep_for(std::chrono::seconds (60 ));
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
}

定时/延时消息收发示例

发送定时/延时消息

#include <iostream>
#include <chrono>
#include <thread>
#include "DefaultMQProducer.h"

using namespace std;
using namespace rocketmq;

int main() {
    std::cout << "=======Before sending messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。
    DefaultMQProducer producer("GID_XXX");
    // 设置为您从消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
    // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。
    producer.setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。
     // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    producer.setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。

    //请确保参数设置完成之后启动Producer。
    producer.start();
    auto start = std::chrono::system_clock::now();
    int count = 32;
    for (int i = 0; i < count; ++i) {
        // 设置为您在消息队列RocketMQ版控制台上创建的定时/延时消息类型的Topic。
        MQMessage msg("YOUR DELAY TOPIC", "HiTAG", "Hello,CPP SDK, Delay Message.");
        chrono::system_clock::duration d = chrono::system_clock::now().time_since_epoch();
        chrono::milliseconds mil = chrono::duration_cast<chrono::milliseconds>(d);
        // 定时延时消息，单位毫秒（ms），在指定时间戳（当前时间之后）进行投递，例如2022-08-07 16:21:00投递。
        // 如果设置成当前时间戳之前的某个时刻，消息将被立刻投递给消费者。
        // 设置需要延时或者定时的时间，例如当前时间延迟10秒后投递。
        long exp = mil.count() + 10000;
        msg.setProperty("__STARTDELIVERTIME", to_string(exp));
        std::cout << "Now: " << mil.count() << " Exp:" << exp << std::endl;
        try {
            SendResult sendResult = producer.send(msg);
            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;
        }
    }
    auto interval = std::chrono::system_clock::now() - start;
    std::cout << "Send " << count << " messages OK, costs "
              << std::chrono::duration_cast<std::chrono::milliseconds>(interval).count() << "ms" << std::endl;

    producer.shutdown();
    std::cout << "=======After sending messages=======" << std::endl;
    return 0;
}

订阅定时/延时消息


#include <iostream>
#include <thread>
#include <chrono>
#include "DefaultMQPushConsumer.h"

using namespace rocketmq;
using namespace std;

class ExampleDelayMessageListener : public MessageListenerConcurrently {
public:
    ConsumeStatus consumeMessage(const std::vector<MQMessageExt> &msgs) {
        for (auto item = msgs.begin(); item != msgs.end(); item++) {
            chrono::system_clock::duration d = chrono::system_clock::now().time_since_epoch();
            chrono::milliseconds mil = chrono::duration_cast<chrono::milliseconds>(d);
            std::cout << "Now: " << mil.count() << " Received Message Topic:" << item->getTopic() << ", MsgId:"
                      << item->getMsgId() << " DelayTime:" << item->getProperty("__STARTDELIVERTIME") << std::endl;
        }
        return CONSUME_SUCCESS;
    }
};

int main(int argc, char *argv[]) {
    std::cout << "=======Before consuming messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。
    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXX");
    // 设置为您从消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。
    // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。
    consumer->setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。
    // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    consumer->setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。
    
    auto start = std::chrono::system_clock::now();

    //register your own listener here to handle the messages received.
    // 请注册自定义侦听函数用来处理接收到的消息，并返回响应的处理结果。
    ExampleDelayMessageListener *messageListener = new ExampleDelayMessageListener();
    consumer->subscribe("YOUR DELAY TOPIC", "*");
    consumer->registerMessageListener(messageListener);

    // 准备工作完成，必须调用启动函数，才可以正常工作。
    // ********************************************
    // 1.确保订阅关系的设置在启动之前完成。
    // 2.确保相同消费者分组内的消费者的订阅关系一致。
    // *********************************************
    consumer->start();

    // 请保持线程常驻，不要执行shutdown操作。
    std::this_thread::sleep_for(std::chrono::seconds(600));
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
}

事务消息

发送事务消息

#include <iostream>
#include <chrono>
#include <thread>
#include "TransactionMQProducer.h"
#include "MQClientException.h"
#include "TransactionListener.h"

using namespace std;
using namespace rocketmq;

class ExampleTransactionListener : public TransactionListener {
public:
    LocalTransactionState executeLocalTransaction(const MQMessage &msg, void *arg) {
        // 执行本地事务，成功返回COMMIT_MESSAGE，失败返回ROLLBACK_MESSAGE，不确定返回UNKNOWN。
        // 如果返回UNKNOWN，则会触发定时任务回查状态。
        std::cout << "Execute Local Transaction,Received Message Topic:" << msg.getTopic() << ", MsgId:"
                  << msg.getBody() << std::endl;
        return UNKNOWN;
    }

    LocalTransactionState checkLocalTransaction(const MQMessageExt &msg) {
        // 回查本地事务执行情况，成功返回COMMIT_MESSAGE，失败返回ROLLBACK_MESSAGE，不确定返回UNKNOWN。
        // 如果返回UNKNOWN，则等待下次定时任务回查。
        std::cout << "Check Local Transaction,Received Message Topic:" << msg.getTopic() << ", MsgId:" << msg.getMsgId()
                  << std::endl;
        return COMMIT_MESSAGE;
    }
};

int main() {
    std::cout << "=======Before sending messages=======" << std::endl;
    // 设置为您在消息队列RocketMQ版控制台创建的Group ID。  
    TransactionMQProducer producer("GID_XXX");
    // 设置为您从消息队列RocketMQ版控制台获取的接入点信息，类似“rmq-cn-XXXX.rmq.aliyuncs.com:8080”。 // 注意！！！直接填写控制台提供的域名和端口即可，请勿添加http://或https://前缀标识，也不要用IP解析地址。producer.setNamesrvAddr("ACCESS POINT");
    // 设置实例用户名、实例密码和用户渠道。用户渠道默认为“ALIYUN”。     
    // 若您使用公网接入消息队列RocketMQ版实例，则必须配置实例的用户名和密码；若使用VPC接入，则无需配置。
    // 如果实例类型为Serverless实例，公网访问必须设置实例的用户名密码，当开启内网免身份识别时,内网访问可以不设置用户名和密码。
    // 实例用户名和密码在消息队列RocketMQ版控制台访问控制的智能身份识别页签中获取。
    producer.setSessionCredentials("INSTANCE USER NAME","INSTANCE PASSWORD","ALIYUN");
    // 注意！！！访问RocketMQ 5.0实例时，InstanceID属性不需要设置，否则会导致失败。

    // 本地事务执行和回查函数。
    ExampleTransactionListener *exampleTransactionListener = new ExampleTransactionListener();
    producer.setTransactionListener(exampleTransactionListener);
    // 请确保参数设置完成之后启动Producer。
    producer.start();
    auto start = std::chrono::system_clock::now();
    int count = 3;
    for (int i = 0; i < count; ++i) {
        // 设置为您在消息队列RocketMQ版控制台上创建的事务消息类型的Topic。
        MQMessage msg("YOUR TRANSACTION TOPIC", "HiTAG", "Hello,CPP SDK, Transaction Message.");
        try {
            SendResult sendResult = producer.sendMessageInTransaction(msg, &i);
            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;
        }
    }
    auto interval = std::chrono::system_clock::now() - start;
    std::cout << "Send " << count << " messages OK, costs "
              << std::chrono::duration_cast<std::chrono::milliseconds>(interval).count() << "ms" << std::endl;

    std::cout << "Wait for local transaction check..... " << std::endl;
    for (int i = 0; i < 6; ++i) {
        this_thread::sleep_for(chrono::seconds(10));
        std::cout << "Running "<< i*10 + 10 << " Seconds......"<< std::endl;
    }
    producer.shutdown();
    std::cout << "=======After sending messages=======" << std::endl;
    return 0;
}

订阅事务消息

订阅事务消息的示例代码和订阅普通消息一样，请参见订阅普通消息。