求.NET 5中使用IAsyncCollector<T>向服务总线写多条消息的Azure Function示例

前置依赖 #

先安装适配.NET 5隔离进程模型的NuGet包：

• Microsoft.Azure.Functions.Worker 版本选1.14.x及以上适配.NET 5的版本
• Microsoft.Azure.Functions.Worker.Extensions.ServiceBus 版本选5.x系列

示例代码 #

以下用HTTP触发作为示例方便测试，你也可以替换为其他触发方式：

using System.Collections.Generic;
using System.Net;
using System.Threading.Tasks;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.Azure.WebJobs.ServiceBus;
using Azure.Messaging.ServiceBus;

public class ServiceBusBatchSendDemo
{
    [Function("BatchSendToServiceBus")]
    public static async Task<HttpResponseData> Run(
        [HttpTrigger(AuthorizationLevel.Function, "post", Route = null)] HttpRequestData req,
        [ServiceBus(
            "<替换为你的服务总线队列/主题名称>", 
            Connection = "ServiceBusConnection", 
            EntityType = ServiceBusEntityType.Queue)] 
        IAsyncCollector<ServiceBusMessage> messageCollector,
        FunctionContext executionContext)
    {
        // 模拟待发送的业务数据列表
        var businessDataList = new List<string>
        {
            "业务数据1",
            "业务数据2",
            "业务数据3",
            "业务数据4",
            "业务数据5"
        };

        // 批量添加消息到收集器
        foreach (var data in businessDataList)
        {
            var serviceBusMsg = new ServiceBusMessage(data);
            // 可选：添加自定义消息属性
            serviceBusMsg.ApplicationProperties.Add("sendSource", "dotnet5-azure-function");
            await messageCollector.AddAsync(serviceBusMsg);
        }

        // 可选：主动刷新缓冲区立即发送所有消息，函数结束前会自动执行该操作
        await messageCollector.FlushAsync();

        var response = req.CreateResponse(HttpStatusCode.OK);
        await response.WriteStringAsync($"已提交{businessDataList.Count}条消息到服务总线");
        return response;
    }
}

本地配置说明 #

在local.settings.json中添加服务总线连接字符串配置：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
    "ServiceBusConnection": "<替换为你的Azure服务总线连接字符串>"
  }
}

注意事项 #

• IAsyncCollector.AddAsync仅将消息写入本地缓冲区，不会立即发送，运行时会自动在缓冲区满或者函数执行结束前批量推送消息到服务总线
• 大批次消息无需手动拆分，服务总线扩展会自动按照单批次最大允许大小/数量做拆分处理
• 如果目标是服务总线主题，只需将EntityType参数值改为ServiceBusEntityType.Topic即可

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