基于EventProcessorClient实现Azure Event Hub无限事件处理可行性咨询

核心问题解答 #

完全可以通过Windows服务的BackgroundService结合EventProcessorClient（或自定义SQL Server检查点存储的EventProcessor<TPartition>）实现无限运行+按事件数触发检查点的需求。EventProcessorClient是Azure Event Hubs官方推荐的长期消费组件，天然支持多分区负载均衡、故障自动恢复、检查点持久化，非常适合Windows服务这类后台长期运行场景。

现有代码问题分析 #

你当前使用EventHubConsumerClient单分区读取的方案存在几个致命问题，导致异常后崩溃终止：

• 无自动重试与故障恢复：ReadEventsFromPartitionAsync一旦因网络中断、Event Hub服务异常抛出未处理异常，外层循环直接终止，没有重连逻辑。
• 未实现检查点：崩溃重启后会从EventPosition.Earliest重新消费，导致数据重复处理。
• 未正确传递终止令牌：ReadEventsFromPartitionAsync没有传入stoppingToken，服务停止时无法优雅终止消费循环。
• 数据库上下文生命周期过长：长期持有同一个IServiceScope和IRepository，可能导致数据库连接泄漏、上下文状态异常。

推荐解决方案：EventProcessorClient + SQL Server检查点存储 #

步骤1：实现SQL Server版CheckpointStore #

自定义实现CheckpointStore接口，将分区所有权、检查点数据存储到SQL Server中，核心逻辑如下：

public class SqlServerCheckpointStore : CheckpointStore
{
    private readonly string _connectionString;

    public SqlServerCheckpointStore(string connectionString)
    {
        _connectionString = connectionString;
    }

    public override async Task<IEnumerable<Checkpoint>> ListCheckpointsAsync(string eventHubName, string consumerGroup, CancellationToken cancellationToken)
    {
        using var conn = new SqlConnection(_connectionString);
        await conn.OpenAsync(cancellationToken);
        var cmd = new SqlCommand("SELECT PartitionId, Offset, SequenceNumber FROM EventHubCheckpoints WHERE EventHubName = @eventHubName AND ConsumerGroup = @consumerGroup", conn);
        cmd.Parameters.AddWithValue("@eventHubName", eventHubName);
        cmd.Parameters.AddWithValue("@consumerGroup", consumerGroup);
        var reader = await cmd.ExecuteReaderAsync(cancellationToken);
        var checkpoints = new List<Checkpoint>();
        while (await reader.ReadAsync(cancellationToken))
        {
            checkpoints.Add(new Checkpoint(
                eventHubName,
                consumerGroup,
                reader.GetString(0),
                reader.GetString(1),
                reader.GetInt64(2)));
        }
        return checkpoints;
    }

    public override async Task UpdateCheckpointAsync(Checkpoint checkpoint, CancellationToken cancellationToken)
    {
        using var conn = new SqlConnection(_connectionString);
        await conn.OpenAsync(cancellationToken);
        var cmd = new SqlCommand(@"
MERGE INTO EventHubCheckpoints AS Target
USING (VALUES (@EventHubName, @ConsumerGroup, @PartitionId, @Offset, @SequenceNumber)) AS Source (EventHubName, ConsumerGroup, PartitionId, Offset, SequenceNumber)
ON Target.EventHubName = Source.EventHubName AND Target.ConsumerGroup = Source.ConsumerGroup AND Target.PartitionId = Source.PartitionId
WHEN MATCHED THEN UPDATE SET Offset = Source.Offset, SequenceNumber = Source.SequenceNumber
WHEN NOT MATCHED THEN INSERT (EventHubName, ConsumerGroup, PartitionId, Offset, SequenceNumber) VALUES (Source.EventHubName, Source.ConsumerGroup, Source.PartitionId, Source.Offset, Source.SequenceNumber);
", conn);
        cmd.Parameters.AddWithValue("@EventHubName", checkpoint.EventHubName);
        cmd.Parameters.AddWithValue("@ConsumerGroup", checkpoint.ConsumerGroup);
        cmd.Parameters.AddWithValue("@PartitionId", checkpoint.PartitionId);
        cmd.Parameters.AddWithValue("@Offset", checkpoint.Offset);
        cmd.Parameters.AddWithValue("@SequenceNumber", checkpoint.SequenceNumber);
        await cmd.ExecuteNonQueryAsync(cancellationToken);
    }

    public override async Task<IEnumerable<PartitionOwnership>> ListOwnershipAsync(string eventHubName, string consumerGroup, CancellationToken cancellationToken)
    {
        using var conn = new SqlConnection(_connectionString);
        await conn.OpenAsync(cancellationToken);
        var cmd = new SqlCommand("SELECT PartitionId, OwnerId, LastModifiedTime FROM EventHubPartitionOwners WHERE EventHubName = @eventHubName AND ConsumerGroup = @consumerGroup", conn);
        cmd.Parameters.AddWithValue("@eventHubName", eventHubName);
        cmd.Parameters.AddWithValue("@consumerGroup", consumerGroup);
        var reader = await cmd.ExecuteReaderAsync(cancellationToken);
        var ownerships = new List<PartitionOwnership>();
        while (await reader.ReadAsync(cancellationToken))
        {
            ownerships.Add(new PartitionOwnership(
                eventHubName,
                consumerGroup,
                reader.GetString(0),
                reader.GetString(1),
                reader.GetDateTimeOffset(2)));
        }
        return ownerships;
    }

    public override async Task<IEnumerable<PartitionOwnership>> ClaimOwnershipAsync(IEnumerable<PartitionOwnership> desiredOwnerships, CancellationToken cancellationToken)
    {
        var owned = new List<PartitionOwnership>();
        foreach (var desired in desiredOwnerships)
        {
            using var conn = new SqlConnection(_connectionString);
            await conn.OpenAsync(cancellationToken);
            var cmd = new SqlCommand(@"
UPDATE EventHubPartitionOwners
SET OwnerId = @NewOwnerId, LastModifiedTime = @NewLastModifiedTime
WHERE EventHubName = @EventHubName AND ConsumerGroup = @ConsumerGroup AND PartitionId = @PartitionId AND (OwnerId IS NULL OR OwnerId = @CurrentOwnerId OR LastModifiedTime < @ExpirationTime)
SELECT @@ROWCOUNT;
", conn);
            cmd.Parameters.AddWithValue("@EventHubName", desired.EventHubName);
            cmd.Parameters.AddWithValue("@ConsumerGroup", desired.ConsumerGroup);
            cmd.Parameters.AddWithValue("@PartitionId", desired.PartitionId);
            cmd.Parameters.AddWithValue("@CurrentOwnerId", desired.OwnerId ?? string.Empty);
            cmd.Parameters.AddWithValue("@NewOwnerId", desired.OwnerId);
            cmd.Parameters.AddWithValue("@NewLastModifiedTime", DateTimeOffset.UtcNow);
            cmd.Parameters.AddWithValue("@ExpirationTime", DateTimeOffset.UtcNow.AddMinutes(-5));
            var rowsAffected = (int)await cmd.ExecuteScalarAsync(cancellationToken);
            if (rowsAffected > 0)
            {
                owned.Add(desired with { LastModifiedTime = DateTimeOffset.UtcNow });
            }
        }
        return owned;
    }
}

步骤2：在BackgroundService中集成EventProcessorClient #

public class LogConsumerBackgroundService : BackgroundService
{
    private const string EventHubConnectionString = "你的Event Hub连接字符串";
    private const string ConsumerGroup = "你的消费组";
    private const string SqlConnectionString = "你的SQL Server连接字符串";
    private const int CheckpointBatchSize = 100; // 每处理100条事件触发检查点

    private readonly ILogger<LogConsumerBackgroundService> _logger;
    private readonly IServiceProvider _serviceProvider;
    private EventProcessorClient _processorClient;
    private readonly Dictionary<string, int> _partitionEventCounts = new Dictionary<string, int>();

    public LogConsumerBackgroundService(ILogger<LogConsumerBackgroundService> logger, IServiceProvider serviceProvider)
    {
        _logger = logger;
        _serviceProvider = serviceProvider;
    }

    public override Task StartAsync(CancellationToken cancellationToken)
    {
        var checkpointStore = new SqlServerCheckpointStore(SqlConnectionString);
        _processorClient = new EventProcessorClient(
            checkpointStore,
            ConsumerGroup,
            EventHubConnectionString);

        _processorClient.ProcessEventAsync += ProcessEventAsync;
        _processorClient.ProcessErrorAsync += ProcessErrorAsync;

        _logger.LogInformation("EventProcessorClient 已初始化");
        return base.StartAsync(cancellationToken);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            await _processorClient.StartProcessingAsync(stoppingToken);
            _logger.LogInformation("事件消费已启动");
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (TaskCanceledException)
        {
            _logger.LogInformation("服务终止信号已收到，正在停止消费");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex.Demystify(), "消费过程中发生未处理异常");
        }
        finally
        {
            await _processorClient.StopProcessingAsync(stoppingToken);
            _logger.LogInformation("事件消费已停止");
        }
    }

    public override void Dispose()
    {
        _processorClient.ProcessEventAsync -= ProcessEventAsync;
        _processorClient.ProcessErrorAsync -= ProcessErrorAsync;
        _processorClient.Dispose();
        base.Dispose();
    }

    private async Task ProcessEventAsync(ProcessEventArgs args)
    {
        try
        {
            using var scope = _serviceProvider.CreateScope();
            var repository = scope.ServiceProvider.GetRequiredService<IRepository>();

            // 解析事件并执行数据库存储逻辑
            var eventData = JsonSerializer.Deserialize<YourEventType>(args.Data.EventBody, new JsonSerializerOptions { PropertyNameCaseInsensitive = true });
            // repository.Add(eventData);
            await repository.SaveChangesAsync(args.CancellationToken);

            // 按事件数触发检查点
            var partitionId = args.Partition.PartitionId;
            lock (_partitionEventCounts)
            {
                _partitionEventCounts.TryGetValue(partitionId, out var count);
                count++;
                _partitionEventCounts[partitionId] = count;

                if (count >= CheckpointBatchSize)
                {
                    _partitionEventCounts[partitionId] = 0;
                    _ = args.UpdateCheckpointAsync(args.CancellationToken);
                    _logger.LogInformation("分区 {PartitionId} 已更新检查点，累计处理 {Count} 条事件", partitionId, CheckpointBatchSize);
                }
            }

            _logger.LogInformation("已处理事件：Partition={PartitionId}, SequenceNumber={SequenceNumber}", partitionId, args.Data.SequenceNumber);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex.Demystify(), "处理事件失败：Partition={PartitionId}, SequenceNumber={SequenceNumber}", args.Partition.PartitionId, args.Data?.SequenceNumber);
            // 业务异常可根据需求选择是否标记检查点（避免重复重试）
            // await args.UpdateCheckpointAsync(args.CancellationToken);
        }
    }

    private Task ProcessErrorAsync(ProcessErrorEventArgs args)
    {
        _logger.LogError(args.Exception.Demystify(), "消费错误：Partition={PartitionId}, ErrorSource={ErrorSource}", args.PartitionId, args.ErrorSource);
        return Task.CompletedTask;
    }
}

关键注意事项 #

• 检查点线程安全：用锁或并发集合统计分区事件数，避免多线程下计数混乱。
• 异常处理策略：业务异常需单独捕获，根据场景决定是否重试或标记检查点；EventProcessorClient会自动处理网络级异常并重连。
• 数据库上下文管理：每次处理事件创建新的IServiceScope，避免长期持有上下文导致的连接泄漏。
• 优雅终止：通过stoppingToken触发服务停止时，StopProcessingAsync会完成当前事件处理并保存检查点。

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