gRPC拦截器中使用async/await实现异步预处理的可行性问题

目前能找到的异步gRPC拦截器示例基本都是后置处理拦截器，比如下面这个错误处理拦截器：

public class ErrorHandlerInterceptor : Interceptor
{
    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
        TRequest request,
        ClientInterceptorContext<TRequest, TResponse> context,
        AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
    {
        var call = continuation(request, context);

        return new AsyncUnaryCall<TResponse>(
            HandleResponse(call.ResponseAsync),
            call.ResponseHeadersAsync,
            call.GetStatus,
            call.GetTrailers,
            call.Dispose);
    }

    private async Task<TResponse> HandleResponse<TResponse>(Task<TResponse> inner)
    {
        try
        {
            return await inner;
        }
        catch (Exception ex)
        {
            throw new InvalidOperationException("Custom error", ex);
        }
    }
}

这类拦截器只能对调用结果做后置处理，没办法在调用发起前修改context。

但如果想要在调用执行前完成异步预处理（比如异步获取元数据并更新请求上下文），像下面这样的实现是否可行？我之前尝试用ContinueWith实现但失败了：

public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
            TRequest request,
            ClientInterceptorContext<TRequest, TResponse> context,
            AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
        {
        var newContext = await PreProcessing(context);
        var call = continuation(request, newContext);

        return new AsyncUnaryCall<TResponse>(
            HandleResponse(call.ResponseAsync),
            call.ResponseHeadersAsync,
            call.GetStatus,
            call.GetTrailers,
            call.Dispose);
    }

private async Task<ClientInterceptorContext<TRequest, TResponse>> PreProcessing(ClientInterceptorContext<TRequest, TResponse> oldContext)
{
    // 执行一些异步操作，比如获取token
    var metaData = await GetAsyncMetadata();
    return new ClientInterceptorContext<TRequest, TResponse>(oldContext.Method, oldContext.Host, oldContext.Options.WithHeaders(metaData));
}

    private async Task<TResponse> HandleResponse<TResponse>(Task<TResponse> inner)
    {
        try
        {
            return await inner;
        }
        catch (Exception ex)
        {
            throw new InvalidOperationException("Custom error", ex);
        }
    }
}

你直接在AsyncUnaryCall方法里用await是行不通的——因为这个方法是同步方法，返回值是AsyncUnaryCall<TResponse>，不支持异步返回。要实现前置异步预处理，核心思路是把异步预处理逻辑嵌入到返回的AsyncUnaryCall的各个异步任务中，让这些任务先完成预处理，再执行实际的gRPC调用。

正确实现示例 #

public class AsyncPreProcessingInterceptor : Interceptor
{
    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
        TRequest request,
        ClientInterceptorContext<TRequest, TResponse> context,
        AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
    {
        // 缓存预处理任务，避免重复执行异步操作
        var preProcessTask = PreProcessing(context);

        async Task<TResponse> ProcessResponseAsync()
        {
            var newContext = await preProcessTask;
            var call = continuation(request, newContext);
            return await HandleResponse(call.ResponseAsync);
        }

        async Task<Metadata> ProcessHeadersAsync()
        {
            var newContext = await preProcessTask;
            var call = continuation(request, newContext);
            return await call.ResponseHeadersAsync;
        }

        async Task<Status> ProcessStatusAsync()
        {
            var newContext = await preProcessTask;
            var call = continuation(request, newContext);
            return await call.GetStatus();
        }

        async Task<Metadata> ProcessTrailersAsync()
        {
            var newContext = await preProcessTask;
            var call = continuation(request, newContext);
            return await call.GetTrailers();
        }

        void DisposeCall()
        {
            // 实际项目需完善：若预处理未完成就触发Dispose，要处理任务取消逻辑
            // 可在PreProcessing中传入CancellationToken，此处为简化示例留空
        }

        return new AsyncUnaryCall<TResponse>(
            ProcessResponseAsync(),
            ProcessHeadersAsync(),
            ProcessStatusAsync,
            ProcessTrailersAsync,
            DisposeCall);
    }

    private async Task<ClientInterceptorContext<TRequest, TResponse>> PreProcessing<TRequest, TResponse>(ClientInterceptorContext<TRequest, TResponse> oldContext)
    {
        // 模拟异步操作：比如从身份服务获取token
        await Task.Delay(100);
        var metaData = new Metadata { { "Authorization", $"Bearer {Guid.NewGuid()}" } };
        return new ClientInterceptorContext<TRequest, TResponse>(oldContext.Method, oldContext.Host, oldContext.Options.WithHeaders(metaData));
    }

    private async Task<TResponse> HandleResponse<TResponse>(Task<TResponse> inner)
    {
        try
        {
            return await inner;
        }
        catch (Exception ex)
        {
            throw new InvalidOperationException("Custom error", ex);
        }
    }
}

关键细节说明 #

1. 避免重复预处理：通过缓存preProcessTask，让ProcessResponseAsync、ProcessHeadersAsync等所有异步任务共享同一个预处理结果，避免重复执行异步操作（比如多次调用身份服务）。
2. Dispose逻辑完善：如果调用方在预处理完成前触发Dispose，需要给PreProcessing方法传入CancellationToken，在DisposeCall中触发取消，防止资源泄漏。
3. 多调用类型适配：上述示例仅针对AsyncUnaryCall，如果要支持流式调用（如AsyncClientStreamingCall、AsyncServerStreamingCall），只需用同样的思路，把异步预处理逻辑嵌入到对应调用类型的异步任务流中即可。

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