如何让带流式响应的C# gRPC ClientInterceptor等待执行完成后退出

你的核心问题是：拦截器里的finally块在continuation返回后立刻执行，导致计时器只统计了发起调用的时间，没覆盖整个异步流的处理时长。因为AsyncServerStreamingCall返回的是包含异步流的对象而非Task，直接await行不通。我们可以通过包装响应流的方式，在流枚举完成（或异常终止）时再记录耗时。

实现思路 #

1. 调用原始continuation拿到原生的AsyncServerStreamingCall对象
2. 包装该对象的ResponseStream，实现自定义的IAsyncEnumerable<TResponse>，在流枚举完成（包括Dispose操作）时停止计时器并输出日志
3. 返回包含包装后流的新AsyncServerStreamingCall对象

完整修改代码 #

首先添加包装异步流的类：

internal class TimedAsyncStream<T> : IAsyncEnumerable<T>, IAsyncDisposable
{
    private readonly IAsyncEnumerable<T> _innerStream;
    private readonly Stopwatch _timer;
    private readonly ILogger _logger;

    public TimedAsyncStream(IAsyncEnumerable<T> innerStream, Stopwatch timer, ILogger logger)
    {
        _innerStream = innerStream;
        _timer = timer;
        _logger = logger;
    }

    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
    {
        try
        {
            await foreach (var item in _innerStream.WithCancellation(cancellationToken).ConfigureAwait(false))
            {
                yield return item;
            }
        }
        finally
        {
            await StopTimerAndLogAsync().ConfigureAwait(false);
        }
    }

    public async ValueTask DisposeAsync()
    {
        if (_innerStream is IAsyncDisposable asyncDisposable)
        {
            await asyncDisposable.DisposeAsync().ConfigureAwait(false);
        }
        await StopTimerAndLogAsync().ConfigureAwait(false);
    }

    private async ValueTask StopTimerAndLogAsync()
    {
        if (!_timer.IsRunning)
            return;
            
        _timer.Stop();
        _logger.LogInformation($"Total elapsed time: {_timer.ElapsedMilliseconds}ms");
        _timer.Reset();
    }
}

然后修改AsyncServerStreamingCall方法：

public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(
    TRequest request,
    ClientInterceptorContext<TRequest, TResponse> context,
    AsyncServerStreamingCallContinuation<TRequest, TResponse> continuation)
{
    LogCall(context.Method);
    AddCallerMetadata(ref context);

    var timer = new Stopwatch();
    timer.Start();

    try
    {
        var originalCall = continuation(request, context);
        // 包装响应流
        var timedStream = new TimedAsyncStream<TResponse>(originalCall.ResponseStream, timer, _logger);
        
        // 返回包装后的AsyncServerStreamingCall
        return new AsyncServerStreamingCall<TResponse>(
            timedStream,
            originalCall.ResponseHeadersAsync,
            originalCall.GetStatus,
            originalCall.GetTrailers,
            originalCall.Dispose);
    }
    catch (Exception ex)
    {
        LogError(ex);
        timer.Stop();
        _logger.LogInformation($"Total elapsed time (failed): {timer.ElapsedMilliseconds}ms");
        throw;
    }
}

关键说明 #

• 包装流会在枚举完成或流被释放时停止计时器，确保统计的是整个异步流从发起到处理完毕的完整时长
• 完全保留原方法签名，符合重写规则，不会破坏拦截器原有逻辑
• 异常分支单独处理计时，避免未记录耗时就抛出异常

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