Todos los productos
Search
Centro de documentos

ApsaraMQ for RocketMQ:Envío y recepción de mensajes programados y diferidos mediante HTTP (C#)

Última actualización: Sep 04, 2026

Este tema proporciona código de ejemplo en C# para enviar y recibir mensajes programados y diferidos mediante el SDK de cliente HTTP de ApsaraMQ for RocketMQ.

Diferencias entre mensajes programados y diferidos

Tanto los mensajes programados como los diferidos posponen la entrega, pero se distinguen en la forma de especificar el momento de envío:

Tipo

Comportamiento de entrega

**Cómo establecer StartDeliverTime**

Mensaje diferido

Se entrega tras un retraso fijo contado desde el momento del envío

Marca de tiempo actual + duración del retraso (en milisegundos)

Mensaje programado

Se entrega en un momento específico

Marca de tiempo objetivo de entrega (en milisegundos)

Mediante HTTP, ambos tipos utilizan la misma propiedad StartDeliverTime, que corresponde a una marca de tiempo absoluta en milisegundos. El broker retiene el mensaje hasta que se alcanza dicha marca de tiempo y luego lo entrega a los consumidores.

Para obtener más información, consulte Scheduled messages and delayed messages.

Casos de uso

  • Tiempo de espera en pedidos: Envíe un mensaje diferido que compruebe el estado del pago después de 30 minutos. Cancele el pedido si no se ha pagado.

  • Reintento con retroceso progresivo: Tras un error transitorio, programe un reintento con un retraso incremental.

  • Notificaciones temporizadas: Entregue un recordatorio en un momento futuro específico.

Requisitos previos

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

Envío de mensajes programados o diferidos

La propiedad StartDeliverTime controla cuándo entrega el broker cada mensaje. Establézcala con una marca de tiempo Unix en milisegundos.

  • Para un mensaje diferido, sume la duración del retraso a la marca de tiempo actual: AliyunSDKUtils.GetNowTimeStamp() + delayMs.

  • Para un mensaje programado, establezca el parámetro con la diferencia de tiempo entre el momento programado y el momento actual.

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

Marcador de posición

Descripción

Ejemplo

<your-http-endpoint>

Endpoint HTTP obtenido de la página Instance Details en la consola de ApsaraMQ for RocketMQ

http://xxxx.mqrest.cn-hangzhou.aliyuncs.com

<your-topic>

Nombre del topic

scheduled-topic

<your-instance-id>

ID de la instancia (establézcalo en null o "" si la instancia no tiene namespace)

MQ_INST_xxxx

using System;
using System.Collections.Generic;
using System.Threading;
using Aliyun.MQ.Model;
using Aliyun.MQ.Model.Exp;
using Aliyun.MQ.Util;

namespace Aliyun.MQ.Sample
{
    public class ProducerSample
    {
        // HTTP endpoint. Get this from the Instance Details page in the ApsaraMQ for RocketMQ console.
        private const string _endpoint = "<your-http-endpoint>";

        // Get AccessKey credentials from environment variables.
        // Do not hardcode credentials in your source code.
        private const string _accessKeyId = Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_ID");
        private const string _secretAccessKey = Environment.GetEnvironmentVariable("ALIBABA_CLOUD_ACCESS_KEY_SECRET");

        // Topic to publish to. Create this in the ApsaraMQ for RocketMQ console.
        private const string _topicName = "<your-topic>";

        // Instance ID. If the instance has no namespace, set this to null or "".
        // Check the Instance Details page to determine whether your instance has a namespace.
        private const string _instanceId = "<your-instance-id>";

        private static MQClient _client = new Aliyun.MQ.MQClient(_accessKeyId, _secretAccessKey, _endpoint);

        static MQProducer producer = _client.GetProducer(_instanceId, _topicName);

        static void Main(string[] args)
        {
            try
            {
                for (int i = 0; i < 4; i++)
                {
                    TopicMessage sendMsg;
                    sendMsg = new TopicMessage("hello mq");

                    // Set a custom property on the message.
                    sendMsg.PutProperty("a", i.ToString());

                    // Key line: set the delivery time.
                    // Delayed message: add the delay (in ms) to the current timestamp.
                    // Scheduled message: set the parameter to the time difference between the scheduled point in time and the current point in time.
                    sendMsg.StartDeliverTime = AliyunSDKUtils.GetNowTimeStamp() + 10 * 1000; // 10-second delay

                    TopicMessage result = producer.PublishMessage(sendMsg);
                    Console.WriteLine("Message published: " + result);
                }
            }
            catch (Exception ex)
            {
                Console.Write(ex);
            }
        }
    }
}

Recepción de mensajes programados o diferidos

Consuma los mensajes programados y diferidos de la misma manera que los mensajes normales. El broker los entrega únicamente después de StartDeliverTime, por lo que no se requiere ninguna configuración especial en el consumidor.

El siguiente ejemplo utiliza sondeo largo (long polling) para consumir mensajes y confirmar cada lote.

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

Marcador de posición

Descripción

Ejemplo

<your-http-endpoint>

Endpoint HTTP obtenido de la página Instance Details

http://xxxx.mqrest.cn-hangzhou.aliyuncs.com

<your-topic>

Nombre del topic

scheduled-topic

<your-instance-id>

ID de la instancia (establézcalo en null o "" si la instancia no tiene namespace)

MQ_INST_xxxx

<your-group-id>

ID del grupo de consumidores

GID_scheduled_consumer

using System;
using System.Collections.Generic;
using System.Threading;
using Aliyun.MQ.Model;
using Aliyun.MQ.Model.Exp;
using Aliyun.MQ;

namespace Aliyun.MQ.Sample
{
    public class ConsumerSample
    {
        // HTTP endpoint. Get this from the Instance Details page in the ApsaraMQ for RocketMQ console.
        private const string _endpoint = "<your-http-endpoint>";

        // The AccessKey ID that is used for authentication.
        private const string _accessKeyId = "${ACCESS_KEY}";
        // The AccessKey secret that is used for authentication.
        private const string _secretAccessKey = "${SECRET_KEY}";

        // Topic to consume from. Create this in the ApsaraMQ for RocketMQ console.
        private const string _topicName = "<your-topic>";

        // Instance ID. If the instance has no namespace, set this to null or "".
        private const string _instanceId = "<your-instance-id>";

        // Consumer group ID. Create this in the ApsaraMQ for RocketMQ console.
        private const string _groupId = "<your-group-id>";

        private static MQClient _client = new Aliyun.MQ.MQClient(_accessKeyId, _secretAccessKey, _endpoint);
        static MQConsumer consumer = _client.GetConsumer(_instanceId, _topicName, _groupId, null);

        static void Main(string[] args)
        {
            // Consume messages in a loop. For production use, run multiple threads concurrently.
            while (true)
            {
                try
                {
                    // Long polling: if no messages are available, the request waits on the broker
                    // for up to the specified number of seconds before returning.
                    List<Message> messages = null;

                    try
                    {
                        messages = consumer.ConsumeMessage(
                            3, // Max messages per batch (up to 16)
                            3  // Long polling timeout in seconds (up to 30)
                        );
                    }
                    catch (Exception exp1)
                    {
                        if (exp1 is MessageNotExistException)
                        {
                            Console.WriteLine(Thread.CurrentThread.Name + " No new message, " + ((MessageNotExistException)exp1).RequestId);
                            continue;
                        }
                        Console.WriteLine(exp1);
                        Thread.Sleep(2000);
                    }

                    if (messages == null)
                    {
                        continue;
                    }

                    List<string> handlers = new List<string>();
                    Console.WriteLine(Thread.CurrentThread.Name + " Receive Messages:");

                    // Process each message.
                    foreach (Message message in messages)
                    {
                        Console.WriteLine(message);
                        Console.WriteLine("Property a is:" + message.GetProperty("a"));
                        handlers.Add(message.ReceiptHandle);
                    }

                    // Acknowledge messages. If the broker does not receive an ACK
                    // before Message.nextConsumeTime, it redelivers the message.
                    // Each delivery assigns a new receipt handle.
                    try
                    {
                        consumer.AckMessage(handlers);
                        Console.WriteLine("Ack message success:");
                        foreach (string handle in handlers)
                        {
                            Console.Write("\t" + handle);
                        }
                        Console.WriteLine();
                    }
                    catch (Exception exp2)
                    {
                        // An ACK can fail if the receipt handle has expired.
                        if (exp2 is AckMessageException)
                        {
                            AckMessageException ackExp = (AckMessageException)exp2;
                            Console.WriteLine("Ack message fail, RequestId:" + ackExp.RequestId);
                            foreach (AckMessageErrorItem errorItem in ackExp.ErrorItems)
                            {
                                Console.WriteLine("\tErrorHandle:" + errorItem.ReceiptHandle + ",ErrorCode:" + errorItem.ErrorCode + ",ErrorMsg:" + errorItem.ErrorMessage);
                            }
                        }
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex);
                    Thread.Sleep(2000);
                }
            }
        }
    }
}

Consulte también