.NET 6.0中如何延迟消费Service Bus消息？

要实现发布消息后延迟指定时长再消费，核心是利用Azure Service Bus原生的延迟消息调度特性，结合环境变量控制延迟时长即可，不需要额外的复杂逻辑。以下是具体实现步骤：

1. 配置延迟时长环境变量 #

先在系统或项目中设置环境变量，比如命名为MESSAGE_DELAY_SECONDS，值为你需要的延迟秒数（比如300代表5分钟）。如果是本地开发，可以在launchSettings.json里配置：

"environmentVariables": {
  "MESSAGE_DELAY_SECONDS": "300"
}

2. 生产者端：发送延迟消息 #

使用Azure官方推荐的Azure.Messaging.ServiceBus SDK（适配.NET 6的版本），在发送消息时通过ScheduledEnqueueTimeUtc指定消息的投递时间——这个时间等于当前时间加上环境变量读取到的延迟时长。

代码示例：

using Azure.Messaging.ServiceBus;
using System;
using System.Threading.Tasks;

public class MessageProducer
{
    private readonly ServiceBusClient _client;
    private readonly ServiceBusSender _sender;

    public MessageProducer(string connectionString, string queueName)
    {
        _client = new ServiceBusClient(connectionString);
        _sender = _client.CreateSender(queueName);
    }

    public async Task SendDelayedMessageAsync(string messageContent)
    {
        // 读取环境变量，未配置则默认延迟60秒
        var delaySeconds = int.TryParse(Environment.GetEnvironmentVariable("MESSAGE_DELAY_SECONDS"), out var seconds) 
            ? seconds 
            : 60;
        var delay = TimeSpan.FromSeconds(delaySeconds);
        var scheduledEnqueueTime = DateTimeOffset.UtcNow.Add(delay);

        // 创建消息并设置延迟投递时间
        var message = new ServiceBusMessage(messageContent)
        {
            ScheduledEnqueueTimeUtc = scheduledEnqueueTime
        };

        await _sender.SendMessageAsync(message);
        Console.WriteLine($"消息已发送，将在 {scheduledEnqueueTime:yyyy-MM-dd HH:mm:ss} UTC 投递");
    }

    public async Task DisposeAsync()
    {
        await _sender.DisposeAsync();
        await _client.DisposeAsync();
    }
}

3. 消费者端：正常监听消息 #

消费者不需要做任何特殊处理，只需要正常监听队列即可。Service Bus会在指定的ScheduledEnqueueTimeUtc时间点，才将消息推送给消费者。

代码示例：

using Azure.Messaging.ServiceBus;
using System;
using System.Threading;
using System.Threading.Tasks;

public class MessageConsumer
{
    private readonly ServiceBusClient _client;
    private readonly ServiceBusProcessor _processor;

    public MessageConsumer(string connectionString, string queueName)
    {
        _client = new ServiceBusClient(connectionString);
        _processor = _client.CreateProcessor(queueName, new ServiceBusProcessorOptions());
        
        // 注册消息处理事件
        _processor.ProcessMessageAsync += ProcessMessageHandler;
        _processor.ProcessErrorAsync += ProcessErrorHandler;
    }

    public async Task StartProcessingAsync(CancellationToken cancellationToken)
    {
        await _processor.StartProcessingAsync(cancellationToken);
        Console.WriteLine("消费者已启动，等待延迟消息投递...");
    }

    private async Task ProcessMessageHandler(ProcessMessageEventArgs args)
    {
        var messageContent = args.Message.Body.ToString();
        Console.WriteLine($"收到延迟消息：{messageContent}，当前时间：{DateTime.UtcNow:yyyy-MM-dd HH:mm:ss} UTC");
        
        // 完成消息处理，标记为已完成
        await args.CompleteMessageAsync(args.Message);
    }

    private Task ProcessErrorHandler(ProcessErrorEventArgs args)
    {
        Console.WriteLine($"消息处理出错：{args.Exception.Message}");
        return Task.CompletedTask;
    }

    public async Task DisposeAsync()
    {
        await _processor.DisposeAsync();
        await _client.DisposeAsync();
    }
}

注意事项 #

• Service Bus的延迟消息最大支持7天的延迟时长，超过这个值会触发报错。
• 如果是需要对已进入队列的消息（比如消费失败后）进行延迟重试，可以使用ScheduleMessageAsync重新调度，但你的场景是发布时直接延迟，上述方法更直接高效。
• 确保使用Azure.Messaging.ServiceBus SDK（版本7.x及以上），旧的Microsoft.Azure.ServiceBus已被弃用，不再推荐使用。

内容的提问来源于stack exchange，提问作者Filipe Eduardo