ApsaraMQ for RocketMQ:通常のメッセージの送受信

前提条件

• C ++ ダイナミックライブラリがインストールされています。 詳細については、「C ++ ダイナミックライブラリのインストール」をご参照ください。

• Alibaba Cloudアカウント用にAccessKeyペアが作成されます。 詳細については、「AccessKey の作成」をご参照ください。

通常のメッセージを送信する

1. 次のコードをProducerDemo.cppファイルにコピーします。 ファイルで、対応するパラメーターを変更し、g ++ コマンドを実行してコードをコンパイルし、実行できる実行可能ファイルを生成します。

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

2. 次のサンプルコードは、Community EditionのTCPクライアントSDK for C ++ を使用して通常のメッセージを送信する方法の例を示しています。

   #include <iostream>
   #include <chrono>
   #include <thread>
   #include "DefaultMQProducer.h"
   
   using namespace std;
   using namespace rocketmq;
   
   int main() {
       std::cout << "=======Before sending messages=======" << std::endl;
       // The ID of the group for which you applied in the ApsaraMQ for RocketMQ console. 
       DefaultMQProducer producer("GID_XXXXXXXXX");
       // The TCP endpoint that you obtained from the Instance Details page in the ApsaraMQ for RocketMQ console. 
       producer.setNamesrvAddr("http://MQ_INST_XXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
       // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
   	  // ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account, which are used for identity verification. 
       // The user channel. Default value: ALIYUN. 
       producer.setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
   
       // After you configure the required parameters, start the producer. 
       producer.start();
       auto start = std::chrono::system_clock::now();
       int count = 32;
       for (int i = 0; i < count; ++i) {
           // Before you send the message, specify the topic for which you applied in the ApsaraMQ for RocketMQ console. 
           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. 次のサンプルコードは、Community EditionのTCPクライアントSDK for C ++ を使用して通常のメッセージを消費する方法の例を示しています。

   #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;
       // The ID of the group for which you applied in the ApsaraMQ for RocketMQ console. 
       DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("GID_XXXXXXXX");
       // The TCP endpoint that you obtained from the Instance Details page in the ApsaraMQ for RocketMQ console. 
       consumer->setNamesrvAddr("http://MQ_INST_XXXXXXXXXXX.mq-internet-access.mq-internet.aliyuncs.com:80");
       // Make sure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are configured. 
   	  // ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account, which are used for identity verification. 
       // The user channel. Default value: ALIYUN. 
       consumer->setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
       auto start = std::chrono::system_clock::now();
   
   
   
       // Use a custom listener function to process the received message and return the result. 
       ExampleMessageListener *messageListener = new ExampleMessageListener();
       consumer->subscribe("YOURTOPIC", "*");
       consumer->registerMessageListener(messageListener);
   
       // The preparation is complete. You must invoke the startup function to start the consumer. 
       // ********************************************
       // 1. Before you start the consumer, make sure that the subscription is configured. 
       // 2. Make sure that the subscriptions of consumers in the same group are consistent. 
       // *********************************************
       consumer->start();
   
       // Keep the main thread running until the process is shut down. 
       std::this_thread::sleep_for(std::chrono::milliseconds(60 * 1000));
       consumer->shutdown();
       std::cout << "=======After consuming messages======" << std::endl;
       return 0;
   }