ApsaraMQ for RocketMQ:Send and receive scheduled and delayed messages

g++ -o delay_producer_demo -std=c++11 -lz -lrocketmq DelayProducerDemo.cpp

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

using namespace std;
using namespace rocketmq;

int main() {
    std::cout << "=======Before sending messages=======" << std::endl;
    // Group ID from the ApsaraMQ for RocketMQ console.
    DefaultMQProducer producer("<group-id>");
    // TCP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console.
    producer.setNamesrvAddr("<endpoint>");
    // Authenticate with your AccessKey ID and AccessKey secret from environment variables.
    // Default user channel: ALIYUN.
    producer.setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");

    // Start the producer.
    producer.start();
    auto start = std::chrono::system_clock::now();
    int count = 32;
    for (int i = 0; i < count; ++i) {
        // Topic from the ApsaraMQ for RocketMQ console.
        MQMessage msg("<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);
        // Set the delivery time.
        // - Scheduled message: use an absolute timestamp in milliseconds.
        //   For example, 1583568060000 delivers the message at 2020-03-07 16:21:00.
        // - Delayed message: add the delay in milliseconds to the current time.
        //   The following example delays delivery by 10 seconds.
        // If the timestamp is earlier than the current time, the message is delivered immediately.
        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;
}

g++ -o delay_consumer_demo -std=c++11 -lz -lrocketmq DelayConsumerDemo.cpp

#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;
    // Group ID from the ApsaraMQ for RocketMQ console.
    DefaultMQPushConsumer *consumer = new DefaultMQPushConsumer("<group-id>");
    // TCP endpoint from the Instance Details page in the ApsaraMQ for RocketMQ console.
    consumer->setNamesrvAddr("<endpoint>");
    // Authenticate with your AccessKey ID and AccessKey secret from environment variables.
    // Default user channel: ALIYUN.
    consumer->setSessionCredentials(getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), "ALIYUN");
    auto start = std::chrono::system_clock::now();

    // Register a message listener.
    ExampleDelayMessageListener *messageListener = new ExampleDelayMessageListener();
    consumer->subscribe("<topic>", "*");
    consumer->registerMessageListener(messageListener);

    // Before starting the consumer:
    // 1. Make sure that the subscription is configured.
    // 2. Make sure that consumers in the same group have consistent subscriptions.
    consumer->start();

    // Keep the thread running. Do not shut down the consumer while messages are being received.
    std::this_thread::sleep_for(std::chrono::seconds(600));
    consumer->shutdown();
    std::cout << "=======After consuming messages======" << std::endl;
    return 0;
}