3. gRPC 拦截器与上下文传播

在分布式事务与消息链路场景中，gRPC 拦截器是承载上下文传播的核心机制。本章聚焦拦截器类型、Metadata 传递、Context 超时与取消传播、链路追踪 ID 注入四个实战维度，帮助你在微服务间建立可靠的事务边界与可观测性基础。

3.1 拦截器类型与职责划分

gRPC 拦截器分为客户端拦截器和服务端拦截器，分别作用于请求发起端和接收端。按调用方式又可细分为一元拦截器（Unary Interceptor）和流式拦截器（Streaming Interceptor）。

3.2 Metadata 传递：事务边界的“信使”

Metadata 是 gRPC 内置的键值对传输机制，用于携带不参与业务序列化的上下文信息。在事务链路中，Metadata 承担以下关键角色：

• 事务 ID 传递：将全局事务标识（如 XID、TraceId）从上游传递到下游服务。
• 超时预算传递：携带剩余超时时间，避免下游服务无限制等待。
• 补偿策略标识：标记事务补偿端点或回滚队列地址。
• 安全凭证：传递 JWT 或临时令牌，确保事务操作权限。

// 客户端拦截器注入 Metadata 示例（Java）
public class ClientTracingInterceptor implements ClientInterceptor {
    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
            MethodDescriptor<ReqT, RespT> method,
            CallOptions callOptions,
            Channel next) {
        return new ForwardingClientCall.SimpleForwardingClientCall<>(next.newCall(method, callOptions)) {
            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                // 注入全局事务 ID 和超时预算
                headers.put(Metadata.Key.of("x-transaction-id", Metadata.ASCII_STRING_MARSHALLER),
                            UUID.randomUUID().toString());
                headers.put(Metadata.Key.of("x-timeout-ms", Metadata.ASCII_STRING_MARSHALLER),
                            String.valueOf(5000));
                super.start(responseListener, headers);
            }
        };
    }
}

3.3 Context 超时与取消传播

gRPC 的 Context 是贯穿 RPC 调用的不可变上下文树，支持超时和取消信号的自动传播。在事务边界中，Context 的超时机制决定了事务的最大等待时间，取消传播则用于快速失败和资源释放。

// 服务端拦截器提取超时并设置 Context（Go 示例）
func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
        // 从 Metadata 提取超时预算
        md, ok := metadata.FromIncomingContext(ctx)
        if ok {
            if timeoutVals := md.Get("x-timeout-ms"); len(timeoutVals) > 0 {
                timeoutMs, err := strconv.Atoi(timeoutVals[0])
                if err == nil {
                    var cancel context.CancelFunc
                    ctx, cancel = context.WithTimeout(ctx, time.Duration(timeoutMs)*time.Millisecond)
                    defer cancel()
                }
            }
        }
        return handler(ctx, req)
    }
}

3.4 链路追踪 ID 注入

在事务边界中，链路追踪 ID（TraceId / SpanId）是串联跨服务调用和Kafka 消息的关键标识。通过拦截器自动注入追踪 ID，可以实现：

• 端到端延迟分析：定位事务在哪个服务或消息队列环节耗时最长。
• 异常根因定位：根据 TraceId 聚合所有相关日志和事件。
• 事务补偿追溯：在补偿操作中携带原始 TraceId，形成完整补偿链路。

// 客户端拦截器注入 TraceId（Java 示例，使用 OpenTelemetry）
public class OpenTelemetryClientInterceptor implements ClientInterceptor {
    private final Tracer tracer;

    public OpenTelemetryClientInterceptor(Tracer tracer) {
        this.tracer = tracer;
    }

    @Override
    public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
            MethodDescriptor<ReqT, RespT> method,
            CallOptions callOptions,
            Channel next) {
        return new ForwardingClientCall.SimpleForwardingClientCall<>(next.newCall(method, callOptions)) {
            @Override
            public void start(Listener<RespT> responseListener, Metadata headers) {
                Span span = tracer.spanBuilder(method.getFullMethodName())
                        .setSpanKind(SpanKind.CLIENT)
                        .startSpan();
                // 将 TraceId 和 SpanId 注入 Metadata
                TextMapPropagator propagator = OpenTelemetry.getGlobalPropagators().getTextMapPropagator();
                propagator.inject(Context.current().with(span), headers, (carrier, key, value) ->
                        headers.put(Metadata.Key.of(key, Metadata.ASCII_STRING_MARSHALLER), value));
                try {
                    super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<>(responseListener) {
                        @Override
                        public void onClose(Status status, Metadata trailers) {
                            span.end();
                            super.onClose(status, trailers);
                        }
                    }, headers);
                } catch (Exception e) {
                    span.recordException(e);
                    span.end();
                    throw e;
                }
            }
        };
    }
}

3.5 参数配置调优清单