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:
C# HTTP istemci SDK'sının yüklü olması. Kurulum talimatları için bkz. Ortamı hazırlama
ApsaraMQ for RocketMQ konsolu içinde bir ApsaraMQ for RocketMQ örneği, konu ve tüketici grubu oluşturulmuş olması
Bir Alibaba Cloud AccessKey çiftinin bulunması. Talimatlar için bkz. AccessKey çifti oluşturma
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.
Planlanmış mesajlar ve gecikmeli mesajlar -- Kavramlar ve sınırlamalar
Kaynak oluşturma -- Örnekleri, konuları ve tüketici gruplarını kurma
Ortamı hazırlama -- C# HTTP istemci SDK'sını yükleme ve yapılandırma