使用EventProcessorClient如何在停机/重启前写入检查点？

核心结论 #

可以在停机时为EventProcessorClient写入最新检查点，关键是通过PartitionClosingAsync事件结合分区最新处理位置的缓存来实现。

具体实现步骤 #

1. 维护线程安全的分区位置缓存
   用ConcurrentDictionary<string, EventPosition>记录每个分区最后成功处理的消息位置，键为分区ID，值为消息的偏移量/序列号信息，确保多线程环境下的安全访问。

2. 在消息处理中更新缓存
   在ProcessEventAsync方法里，每次成功处理消息后，更新对应分区的最新位置到缓存中——不管是否触发了常规的计数器检查点逻辑，这样能保证缓存里始终是当前分区已处理的最新位置。

3. 在PartitionClosingAsync事件中写入最终检查点
   当服务停机触发分区关闭事件时，从缓存中取出该分区的最新位置，通过PartitionContext.UpdateCheckpointAsync方法写入最终检查点。建议仅在正常停机（如Shutdown）场景下执行，避免异常场景下的无效操作。

代码示例 #

// 线程安全的分区最新处理位置缓存
private readonly ConcurrentDictionary<string, EventPosition> _latestProcessedPositions = new();
private int _messageCounter;

public async Task StartProcessing()
{
    var storageClient = new BlobContainerClient("<storage-connection-string>", "<container-name>");
    var processor = new EventProcessorClient(
        storageClient, 
        "<consumer-group>", 
        "<eventhub-connection-string>", 
        "<eventhub-name>");

    // 订阅消息处理和分区关闭事件
    processor.ProcessEventAsync += ProcessEventAsync;
    processor.PartitionClosingAsync += PartitionClosingAsync;

    await processor.StartProcessingAsync();
}

private async Task ProcessEventAsync(ProcessEventArgs args)
{
    try
    {
        // 执行消息处理逻辑
        await HandleMessage(args.EventData);

        // 更新当前分区的最新处理位置
        var latestPosition = EventPosition.FromOffset(args.Data.Offset, isInclusive: false);
        _latestProcessedPositions.AddOrUpdate(
            args.PartitionContext.PartitionId, 
            latestPosition, 
            (_, _) => latestPosition);

        // 原有计数器检查点逻辑：每10条消息写入一次检查点
        if (Interlocked.Increment(ref _messageCounter) % 10 == 0)
        {
            await args.PartitionContext.UpdateCheckpointAsync(args.Data);
        }
    }
    catch (Exception ex)
    {
        // 标记消息处理失败，便于后续重试
        args.FailProcessing(ex);
    }
}

private async Task PartitionClosingAsync(PartitionClosingEventArgs args)
{
    // 仅在正常停机场景下写入最终检查点
    if (args.CloseReason == PartitionCloseReason.Shutdown)
    {
        if (_latestProcessedPositions.TryGetValue(args.PartitionContext.PartitionId, out var latestPosition))
        {
            try
            {
                await args.PartitionContext.UpdateCheckpointAsync(latestPosition);
                Console.WriteLine($"Final checkpoint written for partition {args.PartitionContext.PartitionId}");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Failed to write final checkpoint for partition {args.PartitionContext.PartitionId}: {ex.Message}");
            }
        }
    }
}

注意事项 #

• 缓存仅记录成功处理的消息位置，避免将未处理完成的消息位置作为检查点，导致消息丢失。
• 处理PartitionClosingAsync时要捕获异常，防止检查点写入失败影响服务正常停机流程。
• 若服务是意外崩溃（而非正常调用StopProcessingAsync），PartitionClosingAsync可能无法触发，这种场景下依赖常规的计数器检查点来减少重复消费的范围即可。

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