Tüm Ürünler
Search
Belge Merkezi

ApsaraMQ for RocketMQ:Send and receive scheduled and delayed messages over HTTP (C#)

Son Güncelleme:Oct 04, 2026

Bu konu, ApsaraMQ for RocketMQ HTTP istemci SDK'sı aracılığıyla planlanmış ve gecikmeli mesajların gönderilip alınmasına yönelik C# örnek kodlarını sunar.

Planlanmış ve gecikmeli mesajlar

Planlanmış ve gecikmeli mesajlar teslimatı erteler ancak teslim zamanını belirtme biçimleri farklıdır:

Tür Teslim davranışı **StartDeliverTime ayarlama yöntemi**
Gecikmeli mesaj Gönderim anından itibaren sabit bir gecikme süresi sonunda teslim edilir Geçerli zaman damgası + gecikme süresi (milisaniye cinsinden)
Planlanmış mesaj Belirli bir zaman noktasında teslim edilir Hedef teslim zaman damgası (milisaniye cinsinden)

HTTP üzerinden her iki tür de aynı StartDeliverTime özelliğini kullanır; bu özellik milisaniye cinsinden mutlak bir zaman damgasıdır. Broker, mesajı bu zaman damgasına kadar tutar ve ardından tüketicilere teslim eder.

Daha fazla bilgi için bkz. Planlanmış mesajlar ve gecikmeli mesajlar.

Kullanım senaryoları

  • Sipariş zaman aşımı: Ödeme durumunu 30 dakika sonra kontrol eden gecikmeli bir mesaj gönderin. Ödeme yapılmadıysa siparişi iptal edin.

  • Geri çekilmeli yeniden deneme: Geçici bir arızadan sonra artan bir gecikmeyle yeniden denemeyi planlayın.

  • Zamanlı bildirimler: Belirli bir gelecekteki zamanda hatırlatma gönderin.

Ön koşullar

Başlamadan önce aşağıdakileri tamamladığınızdan emin olun:

Planlanmış veya gecikmeli mesajlar gönderme

StartDeliverTime özelliği, broker'ın her mesajı ne zaman teslim edeceğini kontrol eder. Bu değeri milisaniye cinsinden bir Unix zaman damgası olarak ayarlayın.

  • Gecikmeli mesaj için, geçerli zaman damgasına gecikme süresini ekleyin: AliyunSDKUtils.GetNowTimeStamp() + delayMs.

  • Planlanmış mesaj için, parametreyi planlanan zaman noktası ile geçerli zaman noktası arasındaki zaman farkına ayarlayın.

Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:

Yer Tutucu Açıklama Örnek
<your-http-endpoint> ApsaraMQ for RocketMQ konsolundaki Instance Details sayfasından alınan HTTP uç noktası http://xxxx.mqrest.cn-hangzhou.aliyuncs.com
<your-topic> Konu adı scheduled-topic
<your-instance-id> Örnek Kimliği (örneğin ad alanı yoksa null veya "" olarak ayarlayın) 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);
            }
        }
    }
}

Planlanmış veya gecikmeli mesajlar alma

Planlanmış ve gecikmeli mesajları normal mesajları tükettiğiniz şekilde tüketin. Broker bunları yalnızca StartDeliverTime sonrasında teslim ettiği için özel bir tüketici yapılandırması gerekmez.

Aşağıdaki örnek, uzun oylama kullanarak mesajları tüketir ve her toplu işi onaylar.

Aşağıdaki yer tutucuları gerçek değerlerinizle değiştirin:

Yer Tutucu Açıklama Örnek
<your-http-endpoint> Instance Details sayfasından alınan HTTP uç noktası http://xxxx.mqrest.cn-hangzhou.aliyuncs.com
<your-topic> Konu adı scheduled-topic
<your-instance-id> Örnek Kimliği (örneğin ad alanı yoksa null veya "" olarak ayarlayın) MQ_INST_xxxx
<your-group-id> Tüketici grubu kimliği 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);
                }
            }
        }
    }
}

Ayrıca bkz.