Envío y recepción de mensajes programados y diferidos - ApsaraMQ for RocketMQ

ApsaraMQ for RocketMQ admite dos tipos de entrega de mensajes basada en tiempo: mensajes programados, que se entregan en un momento específico, y mensajes diferidos, que se entregan después de un retraso fijo. Ambos utilizan la propiedad __STARTDELIVERTIME para controlar cuándo el broker entrega el mensaje a los consumidores.

El siguiente código de ejemplo en Java muestra cómo enviar y suscribirse a mensajes programados y diferidos con el SDK de cliente TCP (Community Edition).

Funcionamiento

Cuando un productor envía un mensaje con la propiedad __STARTDELIVERTIME, el broker de ApsaraMQ for RocketMQ retiene el mensaje hasta la hora de entrega especificada. En ese momento, el broker entrega el mensaje a los consumidores suscritos como cualquier mensaje normal.

Mensaje diferido: Establezca __STARTDELIVERTIME en System.currentTimeMillis() + delayMillis. El broker entrega el mensaje una vez que transcurre el retraso.
Mensaje programado: Establezca __STARTDELIVERTIME en una marca de tiempo Unix futura en milisegundos. El broker entrega el mensaje en ese momento exacto.

Si la hora especificada corresponde al pasado, el broker entrega el mensaje de inmediato.

Mensajes programados frente a diferidos: cuándo usar cada uno

Tipo	Caso de uso	Ejemplo
Diferido	Activar una acción tras una espera fija	Cancelar un pedido no pagado 30 minutos después de su creación
Programado	Activar una acción en un momento específico	Enviar una notificación a las 09:00 todos los lunes

Diferencias respecto a Apache RocketMQ

Apache RocketMQ admite mensajes diferidos, pero no es compatible con mensajes programados y carece de una interfaz dedicada para estos últimos.

ApsaraMQ for RocketMQ ofrece capacidades adicionales:

Admite tanto mensajes diferidos como programados
Precisión de segundos para la hora de entrega
Mayor concurrencia en el procesamiento de mensajes basados en tiempo

Importante

Los métodos de configuración y los resultados difieren entre Apache RocketMQ y ApsaraMQ for RocketMQ. Utilice el código de ejemplo de esta página para ApsaraMQ for RocketMQ en la nube.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

La versión 4.5.2 o posterior de la Community Edition del SDK para Java instalada
El entorno preparado
Un par de AccessKey creado para su cuenta de Alibaba Cloud
Un topic y un Group ID creados en la consola de ApsaraMQ for RocketMQ

Envío de mensajes programados y diferidos

Tanto los mensajes programados como los diferidos utilizan la misma propiedad de usuario __STARTDELIVERTIME. La única diferencia radica en cómo se calcula el valor de la marca de tiempo.

Reemplace los siguientes marcadores de posición con sus valores reales:

Marcador de posición	Descripción	Ejemplo
`<your-group-id>`	Group ID creado en la consola de ApsaraMQ for RocketMQ	GID_example
`<your-access-point>`	Endpoint de la instancia obtenido de la consola	`http://MQ_INST_XXXX.aliyuncs.com:80`
`<your-topic>`	Topic creado en la consola de ApsaraMQ for RocketMQ	Topic_example
`<your-message-tag>`	Etiqueta de mensaje para filtrado	TagA

import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class RocketMQProducer {
    /**
     * Reads credentials from environment variables:
     *   ALIBABA_CLOUD_ACCESS_KEY_ID
     *   ALIBABA_CLOUD_ACCESS_KEY_SECRET
     */
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a producer with message trace enabled.
        // To disable message trace, use:
        //   new DefaultMQProducer("<your-group-id>", getAclRPCHook());
        DefaultMQProducer producer = new DefaultMQProducer(
            "<your-group-id>", getAclRPCHook(), true, null);

        // Required for message trace on the cloud.
        producer.setAccessChannel(AccessChannel.CLOUD);

        // Set the instance endpoint from the ApsaraMQ for RocketMQ console.
        producer.setNamesrvAddr("<your-access-point>");
        producer.start();

        for (int i = 0; i < 128; i++) {
            try {
                Message msg = new Message(
                    "<your-topic>",
                    "<your-message-tag>",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));

                // --- Option A: Delayed message ---
                // Deliver 3 seconds from now.
                long delayTime = System.currentTimeMillis() + 3000;
                msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(delayTime));

                // --- Option B: Scheduled message ---
                // Deliver at a specific date and time.
                // Uncomment the following lines and comment out Option A to use.
                //
                // long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
                //     .parse("2025-08-10 18:45:00").getTime();
                // msg.putUserProperty("__STARTDELIVERTIME", String.valueOf(timeStamp));

                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                System.out.println(new Date() + " Send mq message failed.");
                e.printStackTrace();
            }
        }

        // Shut down the producer before exiting (optional).
        producer.shutdown();
    }
}

Resumen:

La propiedad __STARTDELIVERTIME acepta una marca de tiempo Unix en milisegundos.
Para mensajes diferidos, sume el retraso deseado (en milisegundos) a la hora actual.
Para mensajes programados, analice sintácticamente la cadena de fecha y hora objetivo en formato yyyy-MM-dd HH:mm:ss para convertirla en una marca de tiempo Unix.
Si la hora especificada es anterior a la hora actual, el mensaje se entrega de inmediato.

Suscripción a mensajes programados y diferidos

La suscripción a mensajes programados y diferidos es idéntica a la de mensajes normales. El broker retiene el mensaje hasta la hora de entrega especificada, por lo que no se requiere ninguna configuración especial en el lado del consumidor.

import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;

public class RocketMQPushConsumer {
    /**
     * Reads credentials from environment variables:
     *   ALIBABA_CLOUD_ACCESS_KEY_ID
     *   ALIBABA_CLOUD_ACCESS_KEY_SECRET
     */
    private static RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")));
    }

    public static void main(String[] args) throws MQClientException {
        // Create a consumer with message trace enabled.
        // To disable message trace, use:
        //   new DefaultMQPushConsumer("<your-group-id>", getAclRPCHook(),
        //       new AllocateMessageQueueAveragely());
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(
            "<your-group-id>", getAclRPCHook(),
            new AllocateMessageQueueAveragely(), true, null);

        // Set the instance endpoint from the ApsaraMQ for RocketMQ console.
        // The value is in the format of http://xxxx.mq-internet.aliyuncs.com:80.
        consumer.setNamesrvAddr("<your-access-point>");

        // Required for message trace on the cloud.
        consumer.setAccessChannel(AccessChannel.CLOUD);

        // Subscribe to all tags (*) on the topic.
        consumer.subscribe("<your-topic>", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
                    List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
                System.out.printf("Receive New Messages: %s %n", msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
    }
}

Consulte también

Mensajes programados y diferidos: conceptos, precisión de entrega y límites de uso
Explore otros tipos de mensajes, como mensajes transaccionales y mensajes ordenados