如何在Azure EventHub中计算消费者滞后（C# .NET7 Azure Function场景）

一、获取已接收事件的位置（序列号） #

你当前的函数参数用string[] messages只能拿到消息内容，无法获取事件元数据。需要将参数类型改为EventData[]，每个EventData实例会携带事件的核心位置信息：

• SequenceNumber：事件在对应分区的序列号（即已接收事件的位置）
• PartitionId：事件所属的分区ID

修改后的函数代码：

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

[Function("HandleEventHubMessage")]
public async Task Run(
    [EventHubTrigger("<my-eventhub>", Connection = "<MyEventHubConnection>", ConsumerGroup = "<my-consumer-group>")] EventData[] messages, 
    FunctionContext context)
{
    foreach (var eventData in messages)
    {
        // 获取当前事件的序列号和所属分区
        long receivedSequenceNumber = eventData.SequenceNumber;
        string partitionId = eventData.PartitionId;
        
        // 后续计算逻辑...
    }
}

二、获取分区的最后序列号（LastEnqueuedSequenceNumber） #

有两种可靠方式获取：

方式1：从EventData元数据直接读取 #

每个EventData在接收时，会附带Event Hub服务返回的当前分区最新状态，其中包含LastEnqueuedSequenceNumber（该分区已写入的最后一个事件的序列号），无需额外调用即可读取：

foreach (var eventData in messages)
{
    long lastEnqueuedSeqNumber = eventData.LastEnqueuedSequenceNumber;
    long receivedSeqNumber = eventData.SequenceNumber;
    // 计算当前事件对应的滞后值
    long lag = lastEnqueuedSeqNumber - receivedSeqNumber;
}

这种方式性能开销低，适合单事件或批量事件的实时滞后计算。

方式2：通过EventHubConsumerClient主动查询全部分区 #

如果需要批量获取所有分区的最新状态，可创建EventHubConsumerClient查询分区运行时信息：

using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;

// 建议通过依赖注入单例化，避免重复创建客户端
var consumerClient = new EventHubConsumerClient(
    "<my-consumer-group>",
    "<MyEventHubConnection>",
    "<my-eventhub>");

// 获取所有分区的运行时信息
var partitionRuntimeInfos = await consumerClient.GetPartitionRuntimeInformationAsync();
foreach (var partitionInfo in partitionRuntimeInfos)
{
    string partitionId = partitionInfo.PartitionId;
    long lastEnqueuedSeqNumber = partitionInfo.LastEnqueuedSequenceNumber;
    // 结合分区检查点计算整体滞后
}

三、解决BlobCheckpointStore文件为空的问题 #

你遇到的检查点文件为空，通常是以下原因：

1. 未触发首次检查点提交：Azure Functions的EventHubTrigger默认批量处理完成后自动提交检查点，若未处理足够数量的消息（默认批量大小）或未到检查点间隔，不会写入检查点。
2. 检查点配置缺失：可在host.json中显式配置检查点策略，确保及时提交：

{
  "version": "2.0",
  "extensions": {
    "eventHubs": {
      "batchCheckpointFrequency": 1,
      "checkpointMode": "Processed"
    }
  }
}

如果要通过检查点获取消费者组的已确认消费位置，可使用BlobCheckpointStore读取：

using Azure.Messaging.EventHubs.Consumer;
using Azure.Storage.Blobs;

var blobContainerClient = new BlobContainerClient("<MyStorageConnection>", "<checkpoint-container-name>");
var checkpointStore = new BlobCheckpointStore(blobContainerClient);

// 获取消费者组在所有分区的检查点
var checkpoints = await checkpointStore.ListCheckpointsAsync("<my-eventhub>", "<my-consumer-group>");
foreach (var checkpoint in checkpoints)
{
    long checkpointSeqNumber = checkpoint.SequenceNumber;
    string partitionId = checkpoint.PartitionId;
    // 结合分区最后序列号计算滞后
}

四、完整消费者滞后计算逻辑 #

消费者滞后的标准计算方式为：分区最后序列号 - 该分区已消费的最新序列号（已消费序列号可选用检查点记录的确认值，更能反映消费者组的整体进度）

示例代码：

[Function("HandleEventHubMessage")]
public async Task Run(
    [EventHubTrigger("<my-eventhub>", Connection = "<MyEventHubConnection>", ConsumerGroup = "<my-consumer-group>")] EventData[] messages, 
    FunctionContext context)
{
    // 建议通过依赖注入初始化客户端，避免重复创建
    var blobContainerClient = new BlobContainerClient("<MyStorageConnection>", "<checkpoint-container>");
    var checkpointStore = new BlobCheckpointStore(blobContainerClient);
    var consumerClient = new EventHubConsumerClient("<my-consumer-group>", "<MyEventHubConnection>", "<my-eventhub>");

    // 获取所有分区的最新序列号并构建映射
    var partitionInfos = await consumerClient.GetPartitionRuntimeInformationAsync();
    var partitionLastSeqMap = partitionInfos.ToDictionary(p => p.PartitionId, p => p.LastEnqueuedSequenceNumber);

    // 获取消费者组的所有检查点
    var checkpoints = await checkpointStore.ListCheckpointsAsync("<my-eventhub>", "<my-consumer-group>");

    // 计算每个分区的滞后并输出日志
    foreach (var checkpoint in checkpoints)
    {
        if (partitionLastSeqMap.TryGetValue(checkpoint.PartitionId, out var lastSeq))
        {
            long lag = lastSeq - checkpoint.SequenceNumber;
            context.GetLogger("HandleEventHubMessage").LogInformation("分区 {PartitionId} 的消费者滞后: {Lag}", checkpoint.PartitionId, lag);
        }
    }

    // 消息处理逻辑...
}

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