Azure EventHub调用StopProcessingAsync停止时进程挂起问题排查

我使用ASP.NET Core 8 Web API从Azure EventHub读取事件，读取服务托管在HostedService中。处理每个事件前，通过Polly重试机制检查外部API健康状态。当所有重试尝试后API仍不可用时，我尝试停止EventHub处理一段时间，但调用StopProcessingAsync时应用无限挂起，既无异常抛出也无错误日志。

微软文档相关说明：

停止时，处理器会更新其负责处理的分区所有权，并清理与Event Hubs服务通信所用的网络资源。因此，该方法会执行网络I/O，可能需要等待活跃的分区读取完成。

核心问题：

• 调用StopProcessingAsync导致EventHub处理器无限挂起，怀疑是死锁问题
• 想了解如何正确停止EventHub处理并稍后恢复，以及官方推荐的停止方式

附上相关代码：

public class EventHubServiceTask : IEventHubServiceTask
{
    private CancellationToken _startCancellationToken;

    public EventHubServiceTask()
    {
        var storageClient = new BlobContainerClient(_blobStorageSettings.ConnectionString, _blobStorageSettings.ContainerName);
        _eventProcessorClient = new EventProcessorClient(storageClient, EventHubConsumerClient.DefaultConsumerGroupName, _eventHubSettings.ConnectionString, _eventHubSettings.EventHubName);
        _eventProcessorClient.ProcessEventAsync += ProcessEventHandler;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        _startCancellationToken = cancellationToken;
        await _eventProcessorClient.StartProcessingAsync(cancellationToken);
    }

    private async Task ProcessEventHandler(ProcessEventArgs eventArgs)
    {
        MyEvent? myevent= JsonConvert.DeserializeObject<MyEvent>(eventArgs.Data.EventBody.ToString()) ?? new MyEvent();
 
        if (await ServiceHealthCheckNotAvailable(myevent.CorrelationId, eventArgs.CancellationToken))
        {
            await HandleHealthCheckFailureAsync(myevent.CorrelationId, eventArgs.CancellationToken);
        }

        await eventArgs.UpdateCheckpointAsync(eventArgs.CancellationToken);
    }

    private async Task<bool> ServiceHealthCheckNotAvailable(Guid correlationId, CancellationToken cancellationToken)
    {
        return !await CheckHealthAsync(correlationId, cancellationToken);
    }

    private async Task HandleHealthCheckFailureAsync(Guid correlationId, CancellationToken cancellationToken)
    {
        try
        {
            await _eventProcessorClient.StopProcessingAsync(_startCancellationToken);
            await Task.Delay(TimeSpan.FromSeconds(_retryPolicySettings.EventHubSleepTimeSeconds), cancellationToken);
            await _eventProcessorClient.StartProcessingAsync(cancellationToken);
        }
        catch (Exception e)
        {
            throw;
        }
    }

    private async Task<bool> CheckHealthAsync(Guid correlationId, CancellationToken cancellationToken)
    {
        var policy = Policy
                        .HandleResult<bool>(result => !result)
                        .WaitAndRetryAsync(_retryPolicySettings.RetryCount, retryAttempt => TimeSpan.FromSeconds(retryAttempt));

        return await policy.ExecuteAsync(async () => await _service.IsHealthyAsync(correlationId, cancellationToken));
    }
}

死锁原因 #

当前实现存在明确的死锁风险：
ProcessEventHandler由EventProcessorClient的内部线程池调用，在该方法内直接调用StopProcessingAsync时，StopProcessingAsync需要等待所有活跃的事件处理任务完成（包括当前正在执行的ProcessEventHandler），形成循环等待：停止操作等待当前事件处理结束，当前事件处理又等待停止操作完成，最终导致无限挂起。

正确的停止与恢复方式 #

1. 分离停止触发与执行逻辑 #

不要在事件处理线程内直接调用StopProcessingAsync，而是通过独立的信号触发停止，由专门的后台线程处理停止、延迟、重启流程。

修改后的代码示例：

public class EventHubServiceTask : IEventHubServiceTask, IHostedService
{
    private readonly CancellationTokenSource _pauseTokenSource = new CancellationTokenSource();
    private CancellationToken _hostCancellationToken;
    private bool _isPaused;

    public EventHubServiceTask()
    {
        var storageClient = new BlobContainerClient(_blobStorageSettings.ConnectionString, _blobStorageSettings.ContainerName);
        _eventProcessorClient = new EventProcessorClient(storageClient, EventHubConsumerClient.DefaultConsumerGroupName, _eventHubSettings.ConnectionString, _eventHubSettings.EventHubName);
        _eventProcessorClient.ProcessEventAsync += ProcessEventHandler;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        _hostCancellationToken = cancellationToken;
        // 启动EventHub事件处理
        await _eventProcessorClient.StartProcessingAsync(cancellationToken);
        // 启动独立的暂停控制后台任务
        _ = Task.Run(HandlePauseLoopAsync, cancellationToken);
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        await _eventProcessorClient.StopProcessingAsync(cancellationToken);
        _pauseTokenSource.Cancel();
    }

    private async Task ProcessEventHandler(ProcessEventArgs eventArgs)
    {
        if (_isPaused)
        {
            // 暂停状态下可选择跳过或短暂等待，避免占用线程
            await Task.Delay(1000, eventArgs.CancellationToken);
            return;
        }

        MyEvent? myevent= JsonConvert.DeserializeObject<MyEvent>(eventArgs.Data.EventBody.ToString()) ?? new MyEvent();
 
        if (await ServiceHealthCheckNotAvailable(myevent.CorrelationId, eventArgs.CancellationToken))
        {
            // 仅触发暂停信号，不直接处理停止逻辑
            _pauseTokenSource.Cancel();
        }

        await eventArgs.UpdateCheckpointAsync(eventArgs.CancellationToken);
    }

    private async Task HandlePauseLoopAsync()
    {
        while (!_hostCancellationToken.IsCancellationRequested)
        {
            try
            {
                // 等待暂停触发信号
                await _pauseTokenSource.Token.WaitHandle.WaitOneAsync(_hostCancellationToken);
                if (_hostCancellationToken.IsCancellationRequested) break;

                _isPaused = true;
                // 停止EventHub处理，使用主机停止令牌确保响应主机关闭信号
                await _eventProcessorClient.StopProcessingAsync(_hostCancellationToken);
                // 等待指定恢复时间
                await Task.Delay(TimeSpan.FromSeconds(_retryPolicySettings.EventHubSleepTimeSeconds), _hostCancellationToken);
                // 重启EventHub处理
                await _eventProcessorClient.StartProcessingAsync(_hostCancellationToken);
                _isPaused = false;
                // 重置信号源，以便下次触发
                _pauseTokenSource.Dispose();
                _pauseTokenSource = new CancellationTokenSource();
            }
            catch (OperationCanceledException)
            {
                // 主机停止时退出循环
                break;
            }
            catch (Exception ex)
            {
                // 记录异常日志（需注入ILogger）
                // _logger.LogError(ex, "暂停/重启EventHub处理时出错");
                // 异常后延迟重试
                await Task.Delay(5000, _hostCancellationToken);
            }
        }
    }

    // 以下方法保持原逻辑不变
    private async Task<bool> ServiceHealthCheckNotAvailable(Guid correlationId, CancellationToken cancellationToken)
    {
        return !await CheckHealthAsync(correlationId, cancellationToken);
    }

    private async Task<bool> CheckHealthAsync(Guid correlationId, CancellationToken cancellationToken)
    {
        var policy = Policy
                        .HandleResult<bool>(result => !result)
                        .WaitAndRetryAsync(_retryPolicySettings.RetryCount, retryAttempt => TimeSpan.FromSeconds(retryAttempt));

        return await policy.ExecuteAsync(async () => await _service.IsHealthyAsync(correlationId, cancellationToken));
    }
}

2. 关键优化点 #

• 线程分离：事件处理线程仅负责业务逻辑和触发暂停信号，停止/重启逻辑由独立后台线程执行，避免循环等待
• 正确使用令牌：使用HostedService的停止令牌_hostCancellationToken控制停止操作，确保响应主机关闭信号；事件参数中的令牌仅用于取消当前事件处理
• 状态标记：新增_isPaused标记，暂停状态下跳过事件处理，避免无效操作
• 异常处理：在暂停控制循环中捕获异常并添加重试逻辑，提高鲁棒性

3. 官方推荐的停止流程 #

1. 触发暂停信号（通过CancellationTokenSource）
2. 在独立线程中执行：
   • 调用StopProcessingAsync并传入主机停止令牌
   • 等待指定恢复时间
   • 调用StartProcessingAsync重启处理
3. 重置信号源，以便后续再次触发暂停

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