在Azure Function独立进程模式下获取EventHub元数据的方法

问题原因 #

独立进程（Isolated Worker）模式的Azure Functions绑定系统与进程内模式不同，默认会尝试将Event Hub事件内容以JSON格式反序列化到EventData对象，但EventData并非JSON结构，因此抛出序列化异常。直接改用string[]虽然能运行，但会丢失事件元数据。

解决方案 #

方法1：使用Azure.Messaging.EventHubs.EventData类型 #

独立进程模式下，需使用Azure.Messaging.EventHubs命名空间下的EventData类型，配合对应扩展包，即可直接获取完整的事件元数据和内容。

步骤1：安装必要的NuGet包

确保项目中安装以下包：

Install-Package Azure.Messaging.EventHubs
Install-Package Microsoft.Azure.Functions.Worker.Extensions.EventHubs
Install-Package Microsoft.Azure.Functions.Worker.Sdk

步骤2：修改函数代码

注意独立进程模式使用[Function]特性替代原进程内的[FunctionName]，日志通过构造函数注入：

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

public class EventHubProcessor
{
    private readonly ILogger<EventHubProcessor> _logger;

    public EventHubProcessor(ILogger<EventHubProcessor> logger)
    {
        _logger = logger;
    }

    [Function("Function1")]
    public async Task Run([EventHubTrigger("%eventhubname%", Connection = "ehconnection")] EventData[] events)
    {
        foreach (var eventData in events)
        {
            // 获取元数据
            DateTime enqueuedTime = eventData.EnqueuedTime;
            string partitionKey = eventData.PartitionKey;
            long sequenceNumber = eventData.SequenceNumber;
            string offset = eventData.Offset.ToString();

            // 获取事件内容
            string eventContent = eventData.EventBody.ToString();

            _logger.LogInformation($"事件入队时间：{enqueuedTime}，分区键：{partitionKey}，内容：{eventContent}");
            
            // 添加你的事件处理逻辑
        }
    }
}

方法2：绑定特定元数据字段到参数 #

如果只需要部分元数据字段，可以直接将元数据绑定到方法参数，无需使用EventData对象。支持绑定的元数据字段包括：EnqueuedTimeUtc、PartitionKey、Offset、SequenceNumber等，参数需与事件数组长度一致。

示例代码：

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

public class EventHubProcessor
{
    private readonly ILogger<EventHubProcessor> _logger;

    public EventHubProcessor(ILogger<EventHubProcessor> logger)
    {
        _logger = logger;
    }

    [Function("Function1")]
    public async Task Run(
        [EventHubTrigger("%eventhubname%", Connection = "ehconnection")] string[] eventBodies,
        DateTime[] enqueuedTimeUtc,
        string[] partitionKeys,
        long[] sequenceNumbers)
    {
        for (int i = 0; i < eventBodies.Length; i++)
        {
            _logger.LogInformation($"事件{i}：入队时间{enqueuedTimeUtc[i]}，分区键{partitionKeys[i]}，序列号{sequenceNumbers[i]}，内容{eventBodies[i]}");
            
            // 添加你的事件处理逻辑
        }
    }
}

注意事项 #

• 独立进程模式下，函数类需通过构造函数注入ILogger，不能像进程内模式那样将ILogger作为方法参数。
• 确保配置文件（如local.settings.json）中的eventhubname和ehconnection配置项正确。

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