如何通过新版Java客户端动态修改OpenSearch的Apache HttpClient5属性？

问题背景 #

我使用opensearch-java客户端连接AWS上的OpenSearch集群，采用Apache HttpClient5 Transport作为连接方式。当前客户端初始化代码如下：

protected OpenSearchAsyncClient createInstance() throws Exception {
        final HttpHost[] hosts = config.getNodeAddresses().stream().map(
            address -> new HttpHost("https", address, config.getPort())).toArray(HttpHost[]::new);

        final BasicCredentialsProvider credentialsProvider = new BasicCredentialsProvider();

        final SSLContext sslcontext = SSLContextBuilder
                .create()
                .build();

        final ApacheHttpClient5TransportBuilder builder = ApacheHttpClient5TransportBuilder.builder(hosts);
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            final TlsStrategy tlsStrategy = ClientTlsStrategyBuilder.create()
                        .setSslContext(sslcontext)
                        .build();

            final PoolingAsyncClientConnectionManager connectionManager = PoolingAsyncClientConnectionManagerBuilder
                    .create()
                    .setTlsStrategy(tlsStrategy)
                    .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.STRICT)
                    .setMaxConnTotal(config.getMaxConnTotal())
                    .setMaxConnPerRoute(config.getMaxConnPerRoute())
                    .setConnectionTimeToLive(TimeValue.ofSeconds(config.getConnectionTimeToLive()))
                    .setValidateAfterInactivity(TimeValue.ofSeconds(
                            config.getValidateAfterInactivity()))
                    .setConnPoolPolicy(PoolReusePolicy.FIFO)
                    .build();

            return httpClientBuilder
                    .setDefaultCredentialsProvider(credentialsProvider)
                    .setConnectionManager(connectionManager)
                    .evictIdleConnections(TimeValue.ofSeconds(config.getEvictIdleConnections()))
                    .evictExpiredConnections()
                    .setIOReactorConfig(
                            IOReactorConfig.custom()
                                    .setSoTimeout(Timeout.ofSeconds(config.getSoTimeout()))
                                    .setSoKeepAlive(true)
                                    .setIoThreadCount(config.getIoThreadCount())
                                    .setSelectInterval(TimeValue.ofMilliseconds(config.getIOSelectInterval()))
                                    .build())
                    .setIoReactorExceptionCallback(e ->
                            LOGGER.error("OpenSearch client's IOReactor encountered uncaught exception", e));
        });

        builder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                .setConnectionRequestTimeout(Timeout.ofMilliseconds(config.getConnectionRequestTimeout()))
                .setConnectTimeout(Timeout.ofMilliseconds(config.getConnectTimeout()))
                .setResponseTimeout(Timeout.ofMilliseconds(config.getReadTimeout()))
                .setConnectionKeepAlive(TimeValue.ofSeconds(config.getConnectionKeepAlive())));
        builder.setFailureListener(new ApacheHttpClient5Transport.FailureListener() {
            @Override
            public void onFailure(Node node) {
                LOGGER.error("OpenSearch client encountered failure in this node: {}", node.getHost());
            }
        });
        builder.setMapper(new JacksonJsonpMapper(Mapper.getObjectMapper()));

        final OpenSearchTransport transport = builder.build();
        return new OpenSearchAsyncClient(transport);
}

依赖版本 #

<dependency>
    <groupId>org.opensearch.client</groupId>
    <artifactId>opensearch-java</artifactId>
    <version>2.4.0</version>
    <exclusions>
        <exclusion>
            <groupId>commons-logging</groupId>
            <artifactId>commons-logging</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.opensearch</groupId>
    <artifactId>opensearch</artifactId>
    <version>2.5.0</version>
</dependency>
<dependency>
    <groupId>org.apache.httpcomponents.client5</groupId>
    <artifactId>httpclient5</artifactId>
    <version>5.1.4</version>
</dependency>
<dependency>
    <groupId>org.apache.httpcomponents.core5</groupId>
    <artifactId>httpcore5</artifactId>
    <version>5.1.5</version>
</dependency>

需求 #

希望在运行时动态修改客户端属性（如连接池大小、超时时间等），无需修改配置后重启节点，但当前客户端没有公开的动态修改方法，请问能否通过调整配置实现？

解决方案 #

OpenSearch Java客户端本身未提供直接修改已初始化实例属性的API，但可以通过以下几种方式实现动态调整：

1. 配置中心+客户端重建策略

• 监听配置中心的配置变更事件，当配置更新时：
  • 调用旧OpenSearchAsyncClient实例的close()方法，优雅释放资源
  • 使用新配置重新初始化客户端实例
  • 通过AtomicReference<OpenSearchAsyncClient>存储客户端实例，保证切换时的线程安全
• 注意：重建客户端时需确保旧实例的所有请求已完成，避免业务中断

2. 动态调整HttpClient5连接池属性

部分连接池相关属性可通过PoolingAsyncClientConnectionManager的公开方法修改，无需重建客户端：

• 初始化时保留PoolingAsyncClientConnectionManager的类成员引用
• 支持动态修改的属性包括：
  • 最大总连接数：connectionManager.setMaxTotal(int)
  • 单路由最大连接数：connectionManager.setMaxPerRoute(HttpRoute, int)
  • 连接存活时间：connectionManager.setConnectionTimeToLive(long, TimeUnit)
• 示例代码：

// 类成员变量保留连接管理器引用
private PoolingAsyncClientConnectionManager connectionManager;
private HttpHost[] hosts;

protected OpenSearchAsyncClient createInstance() throws Exception {
    hosts = config.getNodeAddresses().stream().map(
        address -> new HttpHost("https", address, config.getPort())).toArray(HttpHost[]::new);
    // ... 其他初始化代码 ...
    this.connectionManager = PoolingAsyncClientConnectionManagerBuilder.create()
            .setTlsStrategy(tlsStrategy)
            .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.STRICT)
            .setMaxConnTotal(config.getMaxConnTotal())
            .setMaxConnPerRoute(config.getMaxConnPerRoute())
            .setConnectionTimeToLive(TimeValue.ofSeconds(config.getConnectionTimeToLive()))
            .setValidateAfterInactivity(TimeValue.ofSeconds(config.getValidateAfterInactivity()))
            .setConnPoolPolicy(PoolReusePolicy.FIFO)
            .build();
    // ... 其他初始化代码 ...
}

// 动态更新连接池配置的方法
public void updateConnectionPoolConfig(int maxConnTotal, int maxConnPerRoute) {
    connectionManager.setMaxTotal(maxConnTotal);
    // 为所有集群节点更新单路由连接数
    for (HttpHost host : hosts) {
        connectionManager.setMaxPerRoute(new HttpRoute(host), maxConnPerRoute);
    }
}

3. 请求级别的配置覆盖

对于超时类属性（连接超时、响应超时等），可以在每次请求时通过RequestOptions动态设置，覆盖客户端全局配置：

SearchRequest searchRequest = new SearchRequest("target_index");
RequestOptions customOptions = RequestOptions.DEFAULT.toBuilder()
        .setConnectTimeout(5000) // 自定义5秒连接超时
        .setSocketTimeout(30000) // 自定义30秒响应超时
        .build();
searchRequest.setOptions(customOptions);
// 发送请求
client.search(searchRequest, customOptions);

注意事项 #

• IO线程数、SSL上下文等涉及底层IO Reactor的配置，无法动态修改，必须重建客户端才能生效
• 客户端重建时需做好资源隔离，避免旧实例的连接池、IO线程等资源泄漏
• 针对AWS OpenSearch集群，重建客户端时要确保IAM签名认证（若使用）配置正确，避免权限异常

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