Azure定时触发Function读取Service Bus主题消息异常问题咨询

问题描述 #

需要定时触发Azure Function（FA），每次运行时读取Azure Service Bus（SB）主题消息，按照文档实现事件处理程序后，FA运行时事件未触发，但相同逻辑在控制台应用中正常运行。使用.NET 6隔离模式FA并部署在Azure服务应用中。

用户代码如下：

using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace FA.Timer.Queue
{
    public class FuncTimerAla
    {
        private readonly ILogger _logger;

        public FuncTimerAla(ILoggerFactory loggerFactory)
        {
            _logger = loggerFactory.CreateLogger<FuncTimerAla>();
        }

        [Function("FuncTimerAla")]
        public async Task Run([TimerTrigger("*/5 * * * * *")] MyInfo myTimer)
        {
            ServiceBusClient client;
            ServiceBusProcessor processor;

            client = new ServiceBusClient("[CONNECTION_STRING]");
            processor = client.CreateProcessor("[TOPIC_NAME]", "[SUBSCRIPTION_NAME]", new ServiceBusProcessorOptions());

            _logger.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");

            try
            {
                processor.ProcessMessageAsync += MessageHandler;
                processor.ProcessErrorAsync += ErrorHandler;

                await processor.StartProcessingAsync();
                _logger.LogInformation($"Wait for a minute and then press any key to end the processor");

                _logger.LogInformation($"Stopping the receiver...");
                await processor.StopProcessingAsync();
                _logger.LogInformation($"Stopped receiving messages");
            }
            catch (Exception ex)
            {
                await processor.DisposeAsync();
                await client.DisposeAsync();
            }
        }

        public async Task MessageHandler(ProcessMessageEventArgs args)
        {
            string body = args.Message.Body.ToString();
            _logger.LogInformation($"Received: {body}");
            await args.CompleteMessageAsync(args.Message);
        }

        public Task ErrorHandler(ProcessErrorEventArgs args)
        {
            Console.WriteLine(args.Exception.ToString());
            return Task.CompletedTask;
        }
    }

    public class MyInfo
    {
        public MyScheduleStatus ScheduleStatus { get; set; }

        public bool IsPastDue { get; set; }
    }

    public class MyScheduleStatus
    {
        public DateTime Last { get; set; }

        public DateTime Next { get; set; }

        public DateTime LastUpdated { get; set; }
    }
}

问题根源 #

核心问题在于定时函数的生命周期与ServiceBusProcessor的工作逻辑不匹配：

• 控制台应用中，启动processor后会等待用户输入才停止，有足够时间接收消息触发事件；
• Azure Function的定时触发函数在Run方法执行完毕后就会结束生命周期，代码中启动processor后立刻调用StopProcessingAsync，processor根本没有时间接收消息，因此MessageHandler事件不会被触发。

解决方案 #

方案一：调整定时函数，预留消息接收时间 #

修改Run方法，在启动processor后等待一段时间，让processor有机会接收并处理消息，再停止处理：

[Function("FuncTimerAla")]
public async Task Run([TimerTrigger("*/5 * * * * *")] MyInfo myTimer)
{
    var client = new ServiceBusClient("[CONNECTION_STRING]");
    var processor = client.CreateProcessor("[TOPIC_NAME]", "[SUBSCRIPTION_NAME]", new ServiceBusProcessorOptions());

    _logger.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");

    var completionSource = new TaskCompletionSource<bool>();
    int processedCount = 0;

    // 重写消息处理逻辑，统计处理数量
    async Task MessageHandler(ProcessMessageEventArgs args)
    {
        string body = args.Message.Body.ToString();
        _logger.LogInformation($"Received: {body}");
        await args.CompleteMessageAsync(args.Message);
        processedCount++;
    }

    // 错误处理时标记任务完成
    Task ErrorHandler(ProcessErrorEventArgs args)
    {
        _logger.LogError(args.Exception, "Service Bus processing error occurred");
        completionSource.TrySetResult(true);
        return Task.CompletedTask;
    }

    processor.ProcessMessageAsync += MessageHandler;
    processor.ProcessErrorAsync += ErrorHandler;
    // 监听processor停止事件，标记任务完成
    processor.Stopped += (s, e) => completionSource.TrySetResult(true);

    await processor.StartProcessingAsync();
    _logger.LogInformation("Started processing messages");

    // 等待10秒让processor接收消息，或直到触发停止条件
    var delayTask = Task.Delay(TimeSpan.FromSeconds(10));
    await Task.WhenAny(delayTask, completionSource.Task);

    await processor.StopProcessingAsync();
    _logger.LogInformation($"Stopped processing, total processed messages: {processedCount}");

    // 释放资源
    await processor.DisposeAsync();
    await client.DisposeAsync();
}

方案二：主动接收消息（更可控的批量处理方式） #

放弃事件模式，使用ServiceBusReceiver主动批量接收消息，更适配定时函数的短生命周期：

[Function("FuncTimerAla")]
public async Task Run([TimerTrigger("*/5 * * * * *")] MyInfo myTimer)
{
    var client = new ServiceBusClient("[CONNECTION_STRING]");
    var receiver = client.CreateReceiver("[TOPIC_NAME]", "[SUBSCRIPTION_NAME]");

    _logger.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");

    try
    {
        // 批量接收消息：最多10条，等待5秒超时
        var messages = await receiver.ReceiveMessagesAsync(maxMessages: 10, maxWaitTime: TimeSpan.FromSeconds(5));

        foreach (var message in messages)
        {
            string body = message.Body.ToString();
            _logger.LogInformation($"Received: {body}");
            // 标记消息为已处理
            await receiver.CompleteMessageAsync(message);
        }

        _logger.LogInformation($"Successfully processed {messages.Count} messages");
    }
    finally
    {
        // 确保资源被释放
        await receiver.DisposeAsync();
        await client.DisposeAsync();
    }
}

额外优化建议 #

• 复用ServiceBusClient：ServiceBusClient是线程安全的，应作为单例注入而非每次函数调用创建，避免连接资源耗尽。在隔离模式的Program.cs中注册：

  builder.Services.AddSingleton(new ServiceBusClient("[CONNECTION_STRING]"));
  

  然后在Function类中注入使用：

  private readonly ServiceBusClient _client;
  private readonly ILogger _logger;
  
  public FuncTimerAla(ServiceBusClient client, ILoggerFactory loggerFactory)
  {
      _client = client;
      _logger = loggerFactory.CreateLogger<FuncTimerAla>();
  }
  

• 优化错误日志：将ErrorHandler中的Console.WriteLine替换为_logger.LogError，让错误日志被Azure Functions日志系统捕获，便于排查问题。

内容的提问来源于stack exchange，提问作者Wulung Triyanto