ApsaraMQ for RocketMQ:Scheduled and delayed messages

Scenarios

Accurate and reliable time-based event triggers are required in scenarios such as distributed timed scheduling and task timeout processing. ApsaraMQ for RocketMQ provides scheduled messages to help you simplify the development of timed scheduling tasks and implement high-performance, scalable, and reliable timed triggering.

Scenario 1: Distributed timed scheduling

A distributed timed scheduling scenario involves tasks that require various time granularity levels. Examples: a task to execute file cleanup at 5 o'clock every day and a task to trigger message pushing every 2 minutes. Traditional dataset-based timed scheduling solutions are complex and inefficient in distributed scenarios. In comparison, scheduled messages in ApsaraMQ for RocketMQ allow you to encapsulate multiple types of time triggers.

Scenario 2: Task timeout processing

A typical scenario that involves task timeout processing is e-commerce payment, where an unpaid order is canceled after it remains unpaid for a specific time period instead of being canceled immediately. In this case, you can use scheduled messages in ApsaraMQ for RocketMQ to check and trigger timeout tasks.

Working mechanism

What is a scheduled message?

Scheduled messages are a type of advance-featured message provided by ApsaraMQ for RocketMQ. After a scheduled message is sent to the broker, the message can be consumed only after a specific period of time or at a specific time. You can use scheduled messages to implement delayed scheduling and triggering in distributed scenarios.

Time setting rules

• The scheduled or delayed time for scheduled messages in ApsaraMQ for RocketMQ is represented as a timestamp, not a time period.

• The scheduled time is in the format of a millisecond-level Unix timestamp. You must convert the scheduled time of message delivery to a millisecond-level Unix timestamp. You can use the Unix timestamp converter to convert a time to a millisecond-level Unix timestamp.

• The scheduled time must be within the allowed time range. If the scheduled time exceeds the range, the scheduled time does not take effect and the messages are immediately delivered by the broker.

• By default, the maximum time range for scheduled messages is 24 hours. You cannot change the default value. For more information, see Limits on parameters.

• The scheduled time must be later than the current time. If the scheduled time is set to a time earlier than the current time, the scheduled time does not take effect and the messages are immediately delivered by the broker.

Lifecycle of a scheduled message

• Initialization

  The message is built and initialized by the producer and is ready to be sent to a broker.

• Being scheduled

  The message is sent to the broker and is stored in a time-based storage system until the specified delivery time is reached. An index is not immediately created for the message.

• Pending consumption

  At the specified time, the message is written into a regular storage engine, where the message is visible for consumers and waits for consumption by consumers.

• Being consumed

  The message is obtained by the consumer and processed based on the local consumption logic of the consumer.

  In this process, the broker waits for the consumer to return the consumption result. If no response is received from the consumer in a specific period of time, ApsaraMQ for RocketMQ performs retries on the message. For more information, see Consumption retry.

• Consumption commit

  The consumer completes the consumption and commits the consumption result to the broker. The broker marks whether the current message is consumed.

  By default, ApsaraMQ for RocketMQ retains all messages. When the consumption result is committed, the message is marked as consumed instead of being immediately deleted. Messages are deleted only if the retention period expires or the system has insufficient storage space. Before a message is deleted, the consumer can re-consume the message.

• Message deletion

  If the retention period of message expires or the storage space is insufficient, ApsaraMQ for RocketMQ deletes the earliest stored messages from the physical file in a rolling manner. For more information, see Message storage and cleanup.

Limits

Message type consistency

Scheduled messages can be sent only to topics whose MessageType is set to Delay.

Time granularity

The scheduled time for scheduled messages in ApsaraMQ for RocketMQ is accurate to milliseconds. The default time granularity value is 1,000 ms.

The status of scheduled messages in ApsaraMQ for RocketMQ can be persistently stored. If the messaging system experiences a failure and is restarted, messages are still delivered based on the specified delivery time. However, if the storage system experiences an exception or is restarted, latency may occur in delivering scheduled messages.

Examples

Unlike normal messages, scheduled messages must have a delivery timestamp specified for them.

The following code provides an example on how to send and receive scheduled messages in Java.

For information about the complete sample code of messaging, see Apache RocketMQ 5.x SDKs (recommend).

        // Send scheduled messages.
        MessageBuilder messageBuilder = new MessageBuilder();
        // Specify a millisecond-level Unix timestamp. In this example, the specified timestamp indicates that the message is delivered in 10 minutes from the current time. 
        Long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;
        Message message = messageBuilder.setTopic("topic")
                // The message key. The system can use the key to locate the message. 
                .setKeys("messageKey")
                // The message tag. The consumer can use the tag to filter messages. 
                .setTag("messageTag")
                .setDeliveryTimestamp(deliverTimeStamp)
                // The message body.
                .setBody("messageBody".getBytes())
                .build();
        try {
            // Send the message. You need to take note of the sending result and capture exceptions such as failures. 
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }

        // Consumption example 1: If a scheduled message is consumed by a push consumer, the consumer needs to process the message only in the message listener. 
        MessageListener messageListener = new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                System.out.println(messageView.getDeliveryTimestamp());
                // Return the status based on the consumption result. 
                return ConsumeResult.SUCCESS;
            }
        };

        // Consumption example 2: If a scheduled message is consumed by a simple consumer, the consumer must obtain the message for consumption and submit the consumption result. 
        List<MessageView> messageViewList = null;
        try {
            messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // After the consumption is complete, the consumer must call the ACK method to commit the consumption result to the broker. 
                try {
                    simpleConsumer.ack(messageView);
                } catch (ClientException e) {
                    e.printStackTrace();
                }
            });
        } catch (ClientException e) {
            // If the message fails to be pulled due to throttling or other reasons, you must re-initiate the request to obtain the message. 
            e.printStackTrace();
        }

Usage notes

We recommend that you do not schedule the same delivery time for a large number of messages.

Scheduled messages are stored in a time-based storage system before they are delivered to consumers at the specified delivery time. If you specify the same delivery time for a large number of scheduled messages, the system has to simultaneously process the messages at the delivery time. This puts the system under heavy load and results in delays in message delivery.

FAQ

Am I able to revoke or change the scheduled time for a scheduled message before the specified scheduled time to send the message is reached?

No, you cannot change the scheduled time for a scheduled message before the specified scheduled time to send the message is reached.

What happens if I specify a scheduled time that is earlier than the current time for a scheduled message?

If you specify a scheduled time that is earlier than the current time for a scheduled message, the scheduled time does not take effect, and the message is immediately delivered.

Why am I unable to query a sent scheduled message in the ApsaraMQ for RocketMQ console?

A scheduled message is visible to consumers and can be queried in the ApsaraMQ for RocketMQ console only after the scheduled time is reached.