Azure Service Bus：迁移至新库后如何持续接收队列消息？

旧库的RegisterMessageHandler在新库中被ServiceBusProcessor替代，这是新库官方推荐的持续接收队列消息的方案，支持并发处理、异常捕获和消息生命周期控制，以下是具体实现逻辑和代码示例：

核心对应关系 #

• 旧库的MessageHandlerOptions对应新库的ServiceBusProcessorOptions，用于配置并发数、自动完成、重试策略等
• 消息处理逻辑绑定到ProcessMessageAsync事件，异常处理绑定到ProcessErrorAsync事件
• 通过StartProcessingAsync()启动持续监听，StopProcessingAsync()停止监听

代码示例 #

旧库（Microsoft.Azure.ServiceBus）实现 #

var queueClient = new QueueClient(connectionString, queueName);
var handlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
{
    MaxConcurrentCalls = 5,
    AutoComplete = true
};
queueClient.RegisterMessageHandler(ProcessMessageAsync, handlerOptions);

async Task ProcessMessageAsync(Message message, CancellationToken token)
{
    // 消息处理逻辑
    Console.WriteLine($"Received message: {Encoding.UTF8.GetString(message.Body)}");
}

Task ExceptionReceivedHandler(ExceptionReceivedEventArgs args)
{
    // 异常处理逻辑
    Console.WriteLine($"Message handler exception: {args.Exception}");
    return Task.CompletedTask;
}

新库（Azure.Messaging.ServiceBus）实现 #

// ServiceBusClient建议单例复用，线程安全
var client = new ServiceBusClient(connectionString);

// 配置处理器选项
var processorOptions = new ServiceBusProcessorOptions
{
    MaxConcurrentCalls = 5,
    AutoCompleteMessages = true // 对应旧库的AutoComplete
};
var processor = client.CreateProcessor(queueName, processorOptions);

// 绑定消息处理逻辑
processor.ProcessMessageAsync += async args =>
{
    // 读取消息内容
    var messageContent = args.Message.Body.ToString();
    Console.WriteLine($"Received message: {messageContent}");
    
    // 若关闭自动完成，需手动调用完成方法
    // await args.CompleteMessageAsync(args.Message);
};

// 绑定异常处理逻辑
processor.ProcessErrorAsync += args =>
{
    Console.WriteLine($"Processor exception: {args.Exception.Message}");
    return Task.CompletedTask;
};

// 启动持续监听
await processor.StartProcessingAsync();

// 停止监听时调用
// await processor.StopProcessingAsync();

关键注意事项 #

• ServiceBusClient是线程安全的，应作为单例在应用中复用，避免频繁创建销毁
• 若需要手动控制消息的完成、放弃或死信，将AutoCompleteMessages设为false，然后调用对应的CompleteMessageAsync、AbandonMessageAsync或DeadLetterMessageAsync方法
• 新库的ServiceBusReceivedMessage替代了旧库的Message，获取内容可通过Body.ToString()（默认UTF-8编码）或Body.ToStream()

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