ApsaraMQ for RocketMQ:Send and subscribe to scheduled messages

Prerequisites

Before you begin, make sure that you have:

• The SDK for C/C++ downloaded and installed

• A development environment configured per Environment preparation (V1.x.x)

• An ApsaraMQ for RocketMQ instance with the required resources (instance, topic, and consumer group)

• An AccessKey pair for your Alibaba Cloud account

Send scheduled messages

The following example sends a scheduled message with a 3-second delivery delay. The setStartDeliverTime method accepts an absolute timestamp in milliseconds. The broker holds the message until that timestamp, then delivers it to consumers.

#include "ONSFactory.h"
#include "ONSClientException.h"

#include <windows.h>
using namespace ons;
int main()
{

    // Create the producer and set the required properties.
    ONSFactoryProperty factoryInfo;
    // The ID of the consumer group that you created in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::ProducerId, "XXX");
    // The TCP endpoint. Find this in the TCP Endpoint section of the Instance Details page in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::NAMESRV_ADDR, "XXX");
    // The topic that you created in the ApsaraMQ for RocketMQ console.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::PublishTopics,"XXX" );
    // The message content.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::MsgContent, "XXX");
    // Make sure the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.
    // AccessKey ID for authentication.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::AccessKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
    // AccessKey secret for authentication.
    factoryInfo.setFactoryProperty(ONSFactoryProperty::SecretKey, getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

    // Create the producer.
    Producer *pProducer = ONSFactory::getInstance()->createProducer(factoryInfo);

    // Call start() once before sending messages.
    pProducer->start();

    Message msg(
            // Message topic.
            factoryInfo.getPublishTopics(),
            // Message tag, used by consumers to filter messages on the broker.
            "TagA",
            // Message body. Must not be empty. The producer and consumer must agree on the serialization and deserialization methods.
            factoryInfo.getMessageContent()
    );

    // The message key is a business-specific identifier. Keep it globally unique
    // so you can query the message by key in the ApsaraMQ for RocketMQ console.
    // Setting a key is optional.
    msg.setKey("ORDERID_100");

    // Set the delivery time as an absolute timestamp in milliseconds.
    // The broker delivers the message to consumers only after this time.
    // In this example, the message is delivered 3 seconds from now.
    long deliverTime = GetTickCount64() + 3000;
    msg.setStartDeliverTime(deliverTime);

    // Send the message. If no exception is thrown, the send succeeded.
    try
    {
        SendResultONS sendResult = pProducer->send(msg);
    }
    catch(ONSClientException & e)
    {
        // Handle the exception.
    }

    // Shut down the producer before exiting to avoid memory leaks.
    pProducer->shutdown();

    return 0;
}

Replace the XXX placeholders with your actual values:

For details on time format rules and limits for setStartDeliverTime, see Scheduled messages and delayed messages.

Subscribe to scheduled messages

Subscribing to scheduled messages works identically to subscribing to normal messages. No additional consumer-side configuration is needed.

For the complete subscription code, see Subscribe to messages.

See also

• Scheduled messages and delayed messages

• Subscribe to messages

• Release notes