.NET8独立工作者模式下Service Bus输出替代IAsyncCollector方案咨询

在独立工作者模式中，IAsyncCollector不再支持直接注入到函数参数用于Service Bus输出，以下是两种可行的替代方案，适配你的_manager.DoStuff需返回Task的要求：

方案一：直接使用Azure.Messaging.ServiceBus客户端手动批量发送 #

直接借助官方ServiceBusClient和ServiceBusSender实现批量逻辑，无需依赖IAsyncCollector。

实现步骤 #

1. 将ServiceBusClient注入到你的业务管理类中
2. 在DoStuff方法内创建队列发送者，构建消息批量并发送

public class MyServiceManager
{
    private readonly ServiceBusClient _serviceBusClient;
    private readonly string _queueName = "your-target-queue";

    // 通过依赖注入传入ServiceBusClient
    public MyServiceManager(ServiceBusClient serviceBusClient)
    {
        _serviceBusClient = serviceBusClient;
    }

    public async Task DoStuff(IEnumerable<string> messageContents)
    {
        using var sender = _serviceBusClient.CreateSender(_queueName);
        using var batch = await sender.CreateBatchAsync();

        foreach (var content in messageContents)
        {
            var message = new ServiceBusMessage(content);
            // 若当前批量无法容纳新消息，先发送现有批量再新建
            if (!batch.TryAddMessage(message))
            {
                await sender.SendMessagesAsync(batch);
                await batch.DisposeAsync();
                
                batch = await sender.CreateBatchAsync();
                batch.TryAddMessage(message);
            }
        }

        // 发送剩余未批量的消息
        if (batch.Count > 0)
        {
            await sender.SendMessagesAsync(batch);
        }
    }
}

方案二：自定义IAsyncCollector实现类 #

如果希望尽量保留原有DoStuff方法的参数签名（依赖IAsyncCollector<string>），可以自定义一个实现IAsyncCollector<string>的类，内部封装Service Bus批量发送逻辑。

自定义Collector实现 #

public class ServiceBusBatchCollector : IAsyncCollector<string>
{
    private readonly ServiceBusSender _sender;
    private ServiceBusMessageBatch _currentBatch;

    public ServiceBusBatchCollector(ServiceBusSender sender)
    {
        _sender = sender;
    }

    public async Task AddAsync(string item, CancellationToken cancellationToken = default)
    {
        _currentBatch ??= await _sender.CreateBatchAsync(cancellationToken);
        
        var message = new ServiceBusMessage(item);
        if (!_currentBatch.TryAddMessage(message))
        {
            // 批量已满，发送后新建
            await _sender.SendMessagesAsync(_currentBatch, cancellationToken);
            _currentBatch = await _sender.CreateBatchAsync(cancellationToken);
            _currentBatch.TryAddMessage(message);
        }
    }

    public async Task FlushAsync(CancellationToken cancellationToken = default)
    {
        // 发送剩余消息
        if (_currentBatch?.Count > 0)
        {
            await _sender.SendMessagesAsync(_currentBatch, cancellationToken);
            await _currentBatch.DisposeAsync();
            _currentBatch = null;
        }
    }
}

在函数中使用自定义Collector #

[Function("YourFunctionName")]
public async Task Run([YourTriggerType] TriggerInput input, ServiceBusClient serviceBusClient)
{
    var sender = serviceBusClient.CreateSender("your-target-queue");
    using var collector = new ServiceBusBatchCollector(sender);

    var manager = new MyServiceManager();
    await manager.DoStuff(collector);

    // 必须手动调用Flush，确保剩余消息被发送
    await collector.FlushAsync();
}

关键注意事项 #

• ServiceBusClient是线程安全的单例对象，建议通过依赖注入以单例模式注入，避免重复创建
• 批量发送需注意Service Bus的单批消息大小限制（默认最大256KB）
• 使用自定义Collector时，必须在调用完DoStuff后手动调用FlushAsync，否则剩余未填满的批量消息不会被发送

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