如何实现非实时读取Azure Service Bus队列？定时触发场景

核心思路 #

放弃实时监听的ServiceBusProcessor，改用ServiceBusReceiver主动发起批量拉取，配合定时触发的Azure Function，在每日结束时间一次性读取队列内所有留存消息。

代码实现 #

定时触发的Azure Function #

using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System.Text.Json;

public class DailyReportFunction
{
    private readonly ServiceBusClient _serviceBusClient;
    private readonly ServiceBusConfig _busConfig;

    // 依赖注入初始化客户端与配置
    public DailyReportFunction(ServiceBusClient serviceBusClient, IOptions<ServiceBusConfig> busConfig)
    {
        _serviceBusClient = serviceBusClient;
        _busConfig = busConfig.Value;
    }

    // 定时触发CRON：每日23:59执行，可根据需求调整
    [Function("DailyReportGenerator")]
    public async Task Run([TimerTrigger("0 59 23 * * *")] TimerInfo timer, ILogger log)
    {
        log.LogInformation("启动批量读取Service Bus队列任务");

        // 创建队列接收器
        using var receiver = _serviceBusClient.CreateReceiver(_busConfig.QueueName);
        var processedNotifications = new List<ChangeNotification>();

        try
        {
            // 批量拉取消息：单次最多拉100条，超时30秒（避免无消息时长期阻塞）
            var messages = await receiver.ReceiveMessagesAsync(maxMessages: 100, timeout: TimeSpan.FromSeconds(30));

            foreach (var msg in messages)
            {
                try
                {
                    // 反序列化消息内容
                    var notification = JsonSerializer.Deserialize<ChangeNotification>(
                        msg.Body,
                        new JsonSerializerOptions { PropertyNameCaseInsensitive = true });

                    processedNotifications.Add(notification);

                    // 处理完成后手动确认，从队列移除消息
                    await receiver.CompleteMessageAsync(msg);
                }
                catch (Exception ex)
                {
                    log.LogError(ex, $"处理消息[{msg.MessageId}]失败");
                    // 处理失败的消息转入死信队列，便于后续排查
                    await receiver.DeadLetterMessageAsync(msg, "处理异常", ex.Message);
                }
            }

            // 执行报告生成逻辑
            await GenerateReport(processedNotifications, log);
        }
        catch (Exception ex)
        {
            log.LogError(ex, "批量读取队列消息时发生全局异常");
        }
        finally
        {
            await receiver.DisposeAsync();
        }
    }

    private async Task GenerateReport(List<ChangeNotification> notifications, ILogger log)
    {
        // 替换为你的报告生成逻辑：写入存储、发送邮件等
        log.LogInformation($"完成{notifications.Count}条消息处理，开始生成当日报告");
        // ...
    }
}

// 配置实体类
public class ServiceBusConfig
{
    public string QueueName { get; set; }
}

// 消息实体类（与你的业务匹配）
public class ChangeNotification
{
    // 你的消息字段定义
    public string Id { get; set; }
    public string Content { get; set; }
    public DateTime CreatedTime { get; set; }
}

关键配置与注意事项 #

• 队列消息留存：确保Service Bus队列的Time To Live设置不小于1天，保证消息能留存到每日触发处理的时间点。
• 批量参数调整：ReceiveMessagesAsync的maxMessages可根据每日预估消息量调整（你的场景设为50即可），timeout避免无消息时函数长期阻塞。
• 消息可靠性：使用手动CompleteMessageAsync确保消息仅在处理成功后被移除；处理失败的消息转入死信队列，避免丢失或重复处理。
• 客户端复用：通过依赖注入管理ServiceBusClient，避免每次触发都创建新连接，提升性能与稳定性。

内容的提问来源于stack exchange，提问作者O'Neil Tomlinson