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 |
|
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:
El SDK de cliente HTTP para C# instalado. Para consultar las instrucciones de configuración, véase Prepare the environment
Una instancia, un topic y un grupo de consumidores de ApsaraMQ for RocketMQ creados en la ApsaraMQ for RocketMQ console
Un par de AccessKey de Alibaba Cloud. Para consultar las instrucciones, véase Create an AccessKey pair
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 |
|
|
Endpoint HTTP obtenido de la página Instance Details en la consola de ApsaraMQ for RocketMQ |
|
|
|
Nombre del topic |
|
|
|
ID de la instancia (establézcalo en |
|
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 |
|
|
Endpoint HTTP obtenido de la página Instance Details |
|
|
|
Nombre del topic |
|
|
|
ID de la instancia (establézcalo en |
|
|
|
ID del grupo de consumidores |
|
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
Scheduled messages and delayed messages -- Conceptos y limitaciones
Create resources -- Configuración de instancias, topics y grupos de consumidores
Prepare the environment -- Instalación y configuración del SDK de cliente HTTP para C#