Este tópico fornece exemplos de código C# para enviar e receber mensagens agendadas e atrasadas usando o SDK do cliente HTTP do ApsaraMQ for RocketMQ.
Mensagens agendadas versus atrasadas
As mensagens agendadas e as atrasadas adiam a entrega, mas diferem na definição do horário:
|
Tipo |
Comportamento de entrega |
**Como definir |
|
Mensagem atrasada |
Entrega após um atraso fixo contado a partir do envio |
Timestamp atual + duração do atraso (em milissegundos) |
|
Mensagem agendada |
Entrega em um momento específico |
Timestamp alvo de entrega (em milissegundos) |
Via HTTP, ambos os tipos usam a propriedade StartDeliverTime, que representa um timestamp absoluto em milissegundos. O broker retém a mensagem até esse instante e depois a entrega aos consumidores.
Para mais informações, consulte Scheduled messages and delayed messages.
Casos de uso
Timeout de pedido: envie uma mensagem atrasada para verificar o status do pagamento após 30 minutos e cancele o pedido se não estiver pago.
Nova tentativa com backoff: após uma falha transitória, agende uma nova tentativa com atraso crescente.
Notificações temporizadas: entregue um lembrete em um horário futuro específico.
Pré-requisitos
Antes de começar, verifique se você tem:
SDK do cliente HTTP C# instalado. Para instruções de configuração, consulte Prepare the environment.
Instância, tópico e grupo de consumidores do ApsaraMQ for RocketMQ criados no console do ApsaraMQ for RocketMQ.
Par de AccessKey da Alibaba Cloud. Para obter instruções, consulte Create an AccessKey pair.
Enviar mensagens agendadas ou atrasadas
A propriedade StartDeliverTime controla quando o broker entrega cada mensagem. Defina-a como um timestamp Unix em milissegundos.
Mensagem atrasada: adicione a duração do atraso ao timestamp atual:
AliyunSDKUtils.GetNowTimeStamp() + delayMs.Mensagem agendada: defina o parâmetro como a diferença entre o momento agendado e o momento atual.
Substitua os seguintes espaços reservados pelos valores reais:
|
Espaço reservado |
Descrição |
Exemplo |
|
|
Endpoint HTTP obtido na página Instance Details no console do ApsaraMQ for RocketMQ |
|
|
|
Nome do tópico |
|
|
|
ID da instância (defina como |
|
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);
}
}
}
}
Receber mensagens agendadas ou atrasadas
Consuma mensagens agendadas e atrasadas da mesma forma que as mensagens normais. O broker as entrega somente após o StartDeliverTime, portanto, nenhuma configuração especial do consumidor é necessária.
O exemplo a seguir usa long polling para consumir mensagens e confirmar cada lote.
Substitua os seguintes espaços reservados pelos valores reais:
|
Espaço reservado |
Descrição |
Exemplo |
|
|
Endpoint HTTP obtido na página Instance Details |
|
|
|
Nome do tópico |
|
|
|
ID da instância (defina como |
|
|
|
ID do 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 também
Scheduled messages and delayed messages — Conceitos e limitações
Create resources — Configure instâncias, tópicos e grupos de consumidores
Prepare the environment — Instale e configure o SDK do cliente HTTP C#