.NET 8+C#应用EventStore订阅频繁断开问题排查与解决

问题背景 #

应用基于C#开发，运行在.NET 8环境，使用EventStore 23.10.0-jammy版本，订阅EventStore时频繁断开，抛出RpcException异常。

异常信息 #

{"Status": {"StatusCode": "Unavailable","Detail": "Error reading next message. HttpIOException: The response ended prematurely while waiting for the next frame from the server. (ResponseEnded)","DebugException":{"HttpRequestError": "ResponseEnded","Message": "The response ended prematurely while waiting for the next frame from the server. (ResponseEnded)","TargetSite": "Void ThrowRequestAborted(System.Exception)","Data": [],"InnerException": null,"HelpLink": null,"Source": "System.Net.Http","HResult": -2146232800,"StackTrace":"   at System.Net.Http.Http2Connection.ThrowRequestAborted(Exception innerException)\r\n   at System.Net.Http.Http2Connection.Http2Stream.CheckResponseBodyState()\r\n   at System.Net.Http.Http2Connection.Http2Stream.TryReadFromBuffer(Span`1 buffer, Boolean partOfSyncRead)\r\n   at System.Net.Http.Http2Connection.Http2Stream.ReadDataAsync(Memory`1 buffer, HttpResponseMessage responseMessage, CancellationToken cancellationToken)\r\n   at Grpc.Net.Client.StreamExtensions.ReadMessageAsync[TResponse](Stream responseStream, GrpcCall call, Func`2 deserializer, String grpcEncoding, Boolean singleMessage, CancellationToken cancellationToken)","$type": "HttpIOException"},"$type": "Status"},"StatusCode": "Unavailable","Trailers": [],"TargetSite": "Void MoveNext()","Message": "Status(StatusCode=\"Unavailable\", Detail=\"Error reading next message. HttpIOException: The response ended prematurely while waiting for the next frame from the server. (ResponseEnded)\", DebugException=\"System.Net.Http.HttpIOException: The response ended prematurely while waiting for the next frame from the server. (ResponseEnded)\r\n   at System.Net.Http.Http2Connection.ThrowRequestAborted(Exception innerException)\r\n   at System.Net.Http.Http2Connection.Http2Stream.CheckResponseBodyState()\r\n   at System.Net.Http.Http2Connection.Http2Stream.TryReadFromBuffer(Span`1 buffer, Boolean partOfSyncRead)\r\n   at System.Net.Http.Http2Connection.Http2Stream.ReadDataAsync(Memory`1 buffer, HttpResponseMessage responseMessage, CancellationToken cancellationToken)\r\n   at Grpc.Net.Client.StreamExtensions.ReadMessageAsync[TResponse](Stream responseStream, GrpcCall call, Func`2 deserializer, String grpcEncoding, Boolean singleMessage, CancellationToken cancellationToken)\")","Data": [], "InnerException": null,"HelpLink": null,"Source": "EventStore.Client","HResult": -2146233088,"StackTrace":"   at EventStore.Client.Interceptors.TypedExceptionInterceptor.AsyncStreamReader`1.MoveNext(CancellationToken cancellationToken)\r\n   at EventStore.Client.AsyncStreamReaderExtensions.ReadAllAsync[T](IAsyncStreamReader`1 reader, CancellationToken cancellationToken)+MoveNext()\r\n   at EventStore.Client.AsyncStreamReaderExtensions.ReadAllAsync[T](IAsyncStreamReader`1 reader, CancellationToken cancellationToken)+System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult()\r\n   at System.Linq.AsyncEnumerable.SelectEnumerableAsyncIterator`2.MoveNextCore() in /_/Ix.NET/Source/System.Linq.Async/System/Linq/Operators/Select.cs:line 221\r\n   at System.Linq.AsyncIteratorBase`1.MoveNextAsync() in /_/Ix.NET/Source/System.Linq.Async/System/Linq/AsyncIterator.cs:line 70\r\n   at System.Linq.AsyncIteratorBase`1.MoveNextAsync() in /_/Ix.NET/Source/System.Linq.Async/System/Linq/AsyncIterator.cs:line 75\r\n   at EventStore.Client.EventStoreClient.ReadInternal(ReadReq request, UserCredentials userCredentials, CancellationToken cancellationToken)+MoveNext()\r\n   at EventStore.Client.EventStoreClient.ReadInternal(ReadReq request, UserCredentials userCredentials, CancellationToken cancellationToken)+MoveNext()\r\n   at EventStore.Client.EventStoreClient.ReadInternal(ReadReq request, UserCredentials userCredentials, CancellationToken cancellationToken)+System.Threading.Tasks.Sources.IValueTaskSource<System.Boolean>.GetResult()\r\n   at EventStore.Client.StreamSubscription.Enumerable.Enumerator.MoveNextAsync()\r\n   at EventStore.Client.StreamSubscription.Subscribe()\r\n   at EventStore.Client.StreamSubscription.Subscribe()","$type": "RpcException"}

订阅代码 #

public async Task<ICustomEventStoreSubscription> SubscribeToEventsAsync(ulong? startFromPosition,Func<EventWithMeta, Task> eventHandler,Action<ICustomEventStoreSubscription,string,Exception?>? droppedHandler)
{
    var startPosition = startFromPosition == null? FromAll.Start: FromAll.After(new Position(startFromPosition.Value, startFromPosition.Value));
    var subscription = await _client.SubscribeToAllAsync(startPosition, EventHandler, subscriptionDropped: DroppedHandler);
    return new CustomEventStoreSubscription(subscription);

    void DroppedHandler(StreamSubscription streamSubscription, SubscriptionDroppedReason resolvedEvent, Exception? cancellationToken)
    {
        droppedHandler?.Invoke(new CustomEventStoreSubscription(streamSubscription), resolvedEvent.ToString(), cancellationToken);
    }

    async Task EventHandler(StreamSubscription streamSubscription, ResolvedEvent resolvedEvent, CancellationToken cancellationToken)
    {
        var eventOrNull = DeserializeEventOrNull(EventTypeAndData.From(resolvedEvent.Event));
        var eventWithMeta = new EventWithMeta(eventOrNull.Item1, eventOrNull.Item2, resolvedEvent.OriginalPosition?.CommitPosition, resolvedEvent.OriginalEventNumber.ToUInt64());
        await eventHandler(eventWithMeta);
    }
}

原因分析 #

• 网络层面问题：订阅基于Http/2长连接，若网络波动、防火墙/负载均衡器超时断开空闲连接，会导致服务器提前关闭响应流，触发该异常。
• 客户端配置缺失：未配置长连接心跳机制，HttpClient未设置合理的存活参数，无法维持连接。
• 事件处理阻塞：EventHandler中处理逻辑耗时过长，导致客户端无法及时响应服务器心跳，引发连接断开。
• EventStore服务端配置：服务器连接超时设置过短，或资源耗尽（CPU/内存过高）主动断开连接。

解决方案 #

1. 配置客户端心跳与长连接参数 #

创建EventStoreClient时，通过SocketsHttpHandler配置心跳和连接存活参数，确保长连接稳定：

var httpHandler = new SocketsHttpHandler
{
    KeepAlivePingDelay = TimeSpan.FromSeconds(30), // 每30秒发送一次心跳
    KeepAlivePingTimeout = TimeSpan.FromSeconds(10), // 心跳超时时间
    PooledConnectionIdleTimeout = TimeSpan.FromMinutes(5), // 连接池空闲超时
    AllowAutoRedirect = false
};

var settings = EventStoreClientSettings.Create("esdb://localhost:2113?tls=false");
settings.HttpMessageHandler = httpHandler;

var client = new EventStoreClient(settings);

2. 实现自动重连机制 #

在订阅断开时，从最后处理的位置自动重连，避免数据丢失：

// 保存最后处理的事件位置
private ulong? _lastProcessedPosition;

public async Task<ICustomEventStoreSubscription> SubscribeToEventsAsync(ulong? startFromPosition,Func<EventWithMeta, Task> eventHandler,Action<ICustomEventStoreSubscription,string,Exception?>? droppedHandler)
{
    var startPosition = startFromPosition ?? _lastProcessedPosition != null 
        ? FromAll.After(new Position(_lastProcessedPosition.Value, _lastProcessedPosition.Value)) 
        : FromAll.Start;
    
    var subscription = await _client.SubscribeToAllAsync(startPosition, EventHandler, subscriptionDropped: DroppedHandler);
    return new CustomEventStoreSubscription(subscription);

    void DroppedHandler(StreamSubscription streamSubscription, SubscriptionDroppedReason reason, Exception? ex)
    {
        droppedHandler?.Invoke(new CustomEventStoreSubscription(streamSubscription), reason.ToString(), ex);
        // 触发重连，添加延迟避免频繁重试
        _ = Task.Run(async () => 
        {
            await Task.Delay(TimeSpan.FromSeconds(5));
            try
            {
                await SubscribeToEventsAsync(_lastProcessedPosition, eventHandler, droppedHandler);
            }
            catch (Exception reconnectEx)
            {
                // 记录重连失败日志
                Console.WriteLine($"重连失败: {reconnectEx.Message}");
                // 递归重试
                _ = ReconnectSubscriptionAsync(eventHandler, droppedHandler);
            }
        });
    }

    async Task EventHandler(StreamSubscription streamSubscription, ResolvedEvent resolvedEvent, CancellationToken cancellationToken)
    {
        var eventOrNull = DeserializeEventOrNull(EventTypeAndData.From(resolvedEvent.Event));
        var eventWithMeta = new EventWithMeta(eventOrNull.Item1, eventOrNull.Item2, resolvedEvent.OriginalPosition?.CommitPosition, resolvedEvent.OriginalEventNumber.ToUInt64());
        await eventHandler(eventWithMeta);
        
        // 更新最后处理的位置
        if (resolvedEvent.OriginalPosition != null)
        {
            _lastProcessedPosition = resolvedEvent.OriginalPosition.Value.CommitPosition;
        }
    }
}

private async Task ReconnectSubscriptionAsync(Func<EventWithMeta, Task> eventHandler, Action<ICustomEventStoreSubscription, string, Exception?>? droppedHandler)
{
    await Task.Delay(TimeSpan.FromSeconds(10)); // 增加重试间隔
    try
    {
        await SubscribeToEventsAsync(_lastProcessedPosition, eventHandler, droppedHandler);
    }
    catch (Exception ex)
    {
        Console.WriteLine($"重连失败: {ex.Message}");
        _ = ReconnectSubscriptionAsync(eventHandler, droppedHandler);
    }
}

3. 优化事件处理逻辑 #

避免在EventHandler中阻塞过长时间，将事件放入异步队列处理，尽快完成订阅回调：

// 注入异步事件队列
private readonly IAsyncQueue<EventWithMeta> _eventQueue;

async Task EventHandler(StreamSubscription streamSubscription, ResolvedEvent resolvedEvent, CancellationToken cancellationToken)
{
    var eventOrNull = DeserializeEventOrNull(EventTypeAndData.From(resolvedEvent.Event));
    var eventWithMeta = new EventWithMeta(eventOrNull.Item1, eventOrNull.Item2, resolvedEvent.OriginalPosition?.CommitPosition, resolvedEvent.OriginalEventNumber.ToUInt64());
    
    // 将事件放入队列异步处理，不阻塞订阅流
    await _eventQueue.EnqueueAsync(eventWithMeta, cancellationToken);
    
    // 立即更新处理位置
    if (resolvedEvent.OriginalPosition != null)
    {
        _lastProcessedPosition = resolvedEvent.OriginalPosition.Value.CommitPosition;
    }
}

4. 检查EventStore服务端配置 #

• 调整EventStore的ConnectionTimeout和KeepAliveInterval配置，延长空闲连接存活时间；
• 监控服务器CPU、内存使用情况，确保资源充足，避免因资源耗尽断开连接。

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