如何为Google Pub/Sub的gRPC调用配置代理凭证？

生效前提 #

首先确保系统环境变量已配置GRPC_PROXY_EXP，值格式为http://<代理地址>:<代理端口>，例如http://127.0.0.1:3128。

核心问题原因 #

你之前的方案未生效主要有两个原因：

• Spring Cloud GCP Pub/Sub底层的gRPC客户端是独立初始化的，不受LogNet、yidongnan等第三方gRPC starter的拦截器管理，因此全局拦截器配置不会生效。
• 你之前给FixedHeaderProvider设置的Proxy-Authorization头只会附加到gRPC业务请求上，代理的CONNECT握手阶段无法读取该头，导致代理认证失败。

正确配置步骤 #

1. 保留原有凭证HTTP传输配置 #

你之前实现的getCredentials()、getHttpTransportFactory()逻辑是正确的，这部分负责Google OAuth鉴权的HTTP请求走代理，需要保留。

2. 自定义gRPC代理认证拦截器 #

专门为GCP Pub/Sub的gRPC通道实现拦截器，在握手阶段注入代理认证头：

class GrpcProxyAuthInterceptor(
    private val proxyUsername: String,
    private val proxyPassword: String
) : ClientInterceptor {
    private val proxyAuthKey = Metadata.Key.of("Proxy-Authorization", Metadata.ASCII_STRING_MARSHALLER)
    private val basicAuth = "Basic " + Base64.getEncoder()
        .encodeToString("$proxyUsername:$proxyPassword".toByteArray())

    override fun <ReqT, RespT> interceptCall(
        method: MethodDescriptor<ReqT, RespT>,
        callOptions: CallOptions,
        next: Channel
    ): ClientCall<ReqT, RespT> {
        return object : SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
            override fun start(responseListener: Listener<RespT>, headers: Metadata) {
                // 握手阶段就注入代理认证头
                headers.put(proxyAuthKey, basicAuth)
                super.start(responseListener, headers)
            }
        }
    }
}

3. 重写SubscriberFactory的通道提供者 #

将自定义拦截器注入到gRPC传输构建逻辑中：

@Bean
fun inboundQuotationsChannelAdapter(
    @Qualifier("inboundQuotationsMessageChannel") quotationsChannel: MessageChannel,
    mpProperties: ConfigurationProperties,
    defaultSubscriberFactory: SubscriberFactory
): PubSubInboundChannelAdapter {
    val proxySubscriberFactory = defaultSubscriberFactory as DefaultSubscriberFactory
    // 保留原有凭证配置，处理OAuth请求的代理逻辑
    proxySubscriberFactory.setCredentialsProvider(ProxyCredentialsProvider(getCredentials()))

    // 构建带代理认证拦截器的gRPC传输提供者
    val proxyAuthInterceptor = GrpcProxyAuthInterceptor(
        proxyUsername = "你的代理用户名",
        proxyPassword = "你的代理密码"
    )
    val transportProvider = SubscriberStubSettings.defaultGrpcTransportProviderBuilder()
        // 注入拦截器，让gRPC CONNECT请求携带代理认证头
        .setInterceptorProvider { listOf(proxyAuthInterceptor) }
        .build()
    proxySubscriberFactory.setChannelProvider(transportProvider)

    val proxySubscriberTemplate = PubSubSubscriberTemplate(proxySubscriberFactory)
    val adapter = PubSubInboundChannelAdapter(proxySubscriberTemplate, mpProperties.gcp.quotationSubscription)
    adapter.outputChannel = quotationsChannel
    adapter.ackMode = AckMode.MANUAL
    adapter.payloadType = ActivityStateChanged::class.java
    return adapter
}

可选优化 #

如果同时需要发送Pub/Sub消息，Publisher的配置逻辑和Subscriber完全一致，找到DefaultPublisherFactory，用同样的方式给它设置带拦截器的TransportProvider即可。

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