收发普通消息

前提条件

• 安装CPP动态库

• 获取阿里云访问密钥AccessKey ID和AccessKey Secret。更多信息，请参见创建AccessKey。

发送普通消息

1. 将以下代码拷贝到ProducerDemo.cpp文件中，修改相应的参数后，使用g++命令进行编译，生成可执行文件，即可直接运行。

   g++ -o producer_demo -std=c++11 -lz -lrocketmq ProducerDemo.cpp

2. 发送普通消息的示例代码如下。

   #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控制台上申请的GID。
       DefaultMQProducer producer("GID_XXXXXXXXX");
       //设置TCP协议接入点，从阿里云RocketMQ控制台的实例详情页面获取。
       producer.setNamesrvAddr("http://MQ_INST_XXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
       //请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
   	  //ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET分别为阿里云账号的AccessKey ID和AccessKey Secret，用于身份验证。
       //用户渠道，默认值为：ALIYUN。
       producer.setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
   
       //请确保参数设置完成之后启动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;
   }

消费普通消息

1. 将以下代码拷贝到ConsumerDemo.cpp文件中，修改相应的参数后，使用g++命令进行编译，生成可执行文件，即可直接运行。

   g++ -o consumer_demo -std=c++11 -lz -lrocketmq ConsumerDemo.cpp

2. 消费普通消息的示例代码如下。

   #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控制台上申请的GID。
       DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXXXXXXX");
       //设置TCP协议接入点，从阿里云消息队列RocketMQ控制台的实例详情页面获取。
       consumer->setNamesrvAddr("http://MQ_INST_XXXXXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
       //请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
   	  //ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET分别为阿里云账号的AccessKey ID和AccessKey Secret，用于身份验证。
       //用户渠道，默认值为：ALIYUN。
       consumer->setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
       auto start = std::chrono::system_clock::now();
   
   
   
       //请注册自定义侦听函数用来处理接收到的消息，并返回响应的处理结果。
       ExampleMessageListener *messageListener = new ExampleMessageListener();
       consumer->subscribe("YOURTOPIC", "*");
       consumer->registerMessageListener(messageListener);
   
       //准备工作完成，必须调用启动函数，才可以正常工作。
       // ********************************************
       // 1.确保订阅关系的设置在启动之前完成。
       // 2.确保相同GID下面的消费者的订阅关系一致。
       // *********************************************
       consumer->start();
   
       // 保持主线程运行直到进程结束。
       std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));
       consumer->shutdown();
       std::cout << "=======After consuming messages======" << std::endl;
       return 0;
   }