Blazor Server中Azure Service Bus Topics消息接收问题求助

问题根源 #

你遇到的情况是Azure Service Bus订阅的竞争消费特性导致的：同一个订阅下的多个处理器会轮询分配消息，加上你用的是单例ServiceBusClient创建的处理器，所有客户端共用同一个处理器实例，自然只会有一个客户端收到消息。而如果尝试为每个客户端创建独立处理器，会变成多处理器竞争订阅，依然是轮着接收消息，达不到全客户端接收的目的。

正确实现思路 #

采用「后台单例服务集中接收消息 + SignalR广播给所有Blazor客户端」的架构：

1. 用单例后台服务作为Service Bus订阅的唯一消费者，接收所有消息
2. 借助Blazor内置的SignalR机制，将消息推送给所有连接的客户端页面

步骤1：创建Service Bus消息接收后台服务 #

public class ServiceBusBroadcastService : IHostedService, IDisposable
{
    private readonly ServiceBusClient _sbClient;
    private readonly IHubContext<ComponentHub> _blazorHubContext;
    private ServiceBusProcessor _processor;

    public ServiceBusBroadcastService(ServiceBusClient sbClient, IHubContext<ComponentHub> blazorHubContext)
    {
        _sbClient = sbClient;
        _blazorHubContext = blazorHubContext;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        // 初始化Processor，配置订阅参数
        _processor = _sbClient.CreateProcessor(
            "你的主题名称", 
            "你的订阅名称", 
            new ServiceBusProcessorOptions
            {
                PrefetchCount = 10,
                ReceiveMode = ServiceBusReceiveMode.PeekLock
            });

        // 注册消息处理逻辑：接收后广播给所有客户端
        _processor.ProcessMessageAsync += async args =>
        {
            var messageContent = args.Message.Body.ToString();
            // 推送消息到所有连接的Blazor客户端
            await _blazorHubContext.Clients.All.SendAsync(
                "ReceiveServiceBusMessage", 
                messageContent, 
                cancellationToken);
            // 标记消息处理完成
            await args.CompleteMessageAsync(args.Message, cancellationToken);
        };

        // 注册错误处理逻辑
        _processor.ProcessErrorAsync += args =>
        {
            // 这里替换成你的日志记录逻辑
            Console.WriteLine($"Service Bus接收错误: {args.Exception.Message}");
            return Task.CompletedTask;
        };

        await _processor.StartProcessingAsync(cancellationToken);
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        if (_processor != null)
        {
            await _processor.StopProcessingAsync(cancellationToken);
            await _processor.DisposeAsync();
        }
    }

    public void Dispose()
    {
        _processor?.Dispose();
    }
}

步骤2：注册服务到依赖注入容器 #

在Program.cs中添加以下配置：

// 注册单例ServiceBusClient（必须单例，避免资源浪费）
builder.Services.AddSingleton<ServiceBusClient>(sp =>
{
    string connectionString = builder.Configuration["ServiceBus:ConnectionString"];
    return new ServiceBusClient(connectionString);
});

// 注册后台消息广播服务
builder.Services.AddHostedService<ServiceBusBroadcastService>();

// 注册Blazor内置的SignalR Hub支持
builder.Services.AddSignalR();

步骤3：在Blazor页面中订阅消息推送 #

@page "/message-receiver"
@inject IHubContext<ComponentHub> BlazorHubContext
@implements IAsyncDisposable

<h3>实时消息接收</h3>
<div class="message-list">
    @foreach (var msg in _receivedMessages)
    {
        <p>@msg</p>
    }
</div>

@code {
    private List<string> _receivedMessages = new();
    private IDisposable _messageSubscription;

    protected override async Task OnInitializedAsync()
    {
        // 订阅后台服务推送的消息
        _messageSubscription = BlazorHubContext.Clients.All.On<string>(
            "ReceiveServiceBusMessage", 
            message =>
            {
                _receivedMessages.Add(message);
                // 通知组件更新UI
                StateHasChanged();
            });
        
        await base.OnInitializedAsync();
    }

    // 页面销毁时取消订阅
    public async ValueTask DisposeAsync()
    {
        _messageSubscription?.Dispose();
    }
}

关键说明 #

• 避免多Processor竞争：后台服务作为唯一消费者，确保所有消息都被接收，再通过SignalR广播，每个客户端都能收到完整的消息流
• ServiceBusClient单例复用：该对象是重量级资源，单例注册能大幅提升性能、减少资源消耗
• 消息可靠性：采用PeekLock接收模式，确保消息处理完成后才标记为已消费，避免丢失

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