从 Java 反应器上下文创建/重新创建 Azure CosmosDb 异步客户端

Question

根据公司政策，我们的 CosmosDb 密钥会与 Azure KeyVault 中的新密钥定期轮换，因此我们需要在 Java weblfux/reactive 应用程序中处理此更改的密钥。从 Java SDK 中，一旦从主密钥创建了 CosmosAsyncClient，就无法更改它，我们需要使用大致以下代码重建它：

SecretAsyncClient secretAsyncClient = new SecretClientBuilder().buildAsyncClient()
...
Mono<CosmosAsyncClient> client = secretAsyncClient.getSecret(KEY_NAME).map(
  s -> s.getValue()
).map(
  key -> new CosmosClientBuilder()
                        .endpoint(HOST)
                        .key(key)
                        .buildAsyncClient()
);
return client.flatMap(
   ...
);

从日志中，

getSecret()

没问题，但后来我得到：

05:25:00.597 INFO  c.a.c.i.RxDocumentClientImpl - Initializing DocumentClient [4] with serviceEndpoint [https://xxxx-cosmosdb-sql-dev.documents.azure.com:443/], connectionPolicy [ConnectionPolicy{httpNetworkRequestTimeout=PT1M, tcpNetworkRequestTimeout=PT5S, connectionMode=DIRECT, maxConnectionPoolSize=1000, idleHttpConnectionTimeout=PT1M, idleTcpConnectionTimeout=PT0S, userAgentSuffix='', throttlingRetryOptions=RetryOptions{maxRetryAttemptsOnThrottledRequests=9, maxRetryWaitTime=PT30S}, endpointDiscoveryEnabled=true, preferredRegions=[Switzerland North], multipleWriteRegionsEnabled=true, proxyType=null, inetSocketProxyAddress=null, readRequestsFallbackEnabled=true, connectTimeout=PT5S, idleTcpEndpointTimeout=PT1H, maxConnectionsPerEndpoint=130, maxRequestsPerConnection=30, tcpConnectionEndpointRediscoveryEnabled=true}], consistencyLevel [Session], directModeProtocol [Tcp]05:25:00.598 ERROR c.a.c.i.RxDocumentClientImpl - unexpected failure in initializing client.
java.lang.RuntimeException: java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-epoll-3
        at com.azure.cosmos.implementation.http.ReactorNettyClient.attemptToWarmupHttpClient(ReactorNettyClient.java:118)
        at com.azure.cosmos.implementation.http.ReactorNettyClient.createWithConnectionProvider(ReactorNettyClient.java:98)
        at com.azure.cosmos.implementation.http.HttpClient.createFixed(HttpClient.java:61)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.httpClient(RxDocumentClientImpl.java:617)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.<init>(RxDocumentClientImpl.java:410)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.<init>(RxDocumentClientImpl.java:262)
        at com.azure.cosmos.implementation.RxDocumentClientImpl.<init>(RxDocumentClientImpl.java:230)
        at com.azure.cosmos.implementation.AsyncDocumentClient$Builder.build(AsyncDocumentClient.java:243)
        at com.azure.cosmos.CosmosAsyncClient.<init>(CosmosAsyncClient.java:129)
        at com.azure.cosmos.CosmosClientBuilder.buildAsyncClient(CosmosClientBuilder.java:779)

所以看来我需要将

CosmosClientBuilder().buildAsyncClient()

部分放入执行器池中？有没有更简单或更优雅的方法来做到这一点？

Answer 1

我对源代码进行了简短的挖掘，错误消息是正确的。 CosmosClientAsync 中有一个

阻塞呼叫

。

该库做了一些实际上相当丑陋的事情，他们对 Netty HttpClient 中的 warmup 函数进行了

动态查找

，然后调用它。

Netty HttpClient 上的预热功能（根据文档）执行以下操作：

根据实际配置，返回一个触发的Mono：

• 事件循环组的初始化
• 主机名解析器的初始化
• 加载传输所需的本机库

默认情况下，当不使用方法时，连接操作会吸收初始化和加载资源所需的额外时间。

在 Cosmos 库中，此预热调用似乎不可配置。

我想我实际上会在他们的 github 中打开一个问题并向他们询问这个问题。

Answer 2

正如我在这里所解释的，由于分布式系统的性质，Cosmos DB SDK 客户端会创建大量连接。 Warmup 是一个需要在客户端创建时实现的功能。

关于您最初的旋转密钥问题，我在这里提供了详细信息 - https://github.com/Azure/azure-sdk-for-java/issues/40051#issuecomment-2097118239

您可以使用 AzureKeyCredential 接口，该接口允许轮换密钥，而无需每次都重新创建客户端。 https://azuresdkdocs.blob.core.windows.net/$web/java/azure-cosmos/latest/com/azure/cosmos/CosmosClientBuilder.html#credential(com.azure.core.credential.AzureKeyCredential)