Java AWS SDK 2.x 异步客户端 - 无法覆盖线程池执行器

Question

我想为 AWS SDK 2.x 设置异步客户端的自定义线程池执行程序(例如 SnsAsyncClient)。我发现对于 AWS SDK 1.x，客户端生成器上存在 withExecutorFactory，对于 2.x，我们需要设置 ClientAsyncConfiguration 并调用 asyncConfiguration 客户端构建器上的方法:

ClientAsyncConfiguration.Builder asyncConfig = ClientAsyncConfiguration.builder()
        .advancedOptions(Map.of(SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR,
                myCustomExecutor));
SnsAsyncClient.builder()
        .asyncConfiguration(asyncConfig.build())
        .build();

不过，自定义线程池执行器不适用。我希望对 CompletableFuture 的所有异步调用和后续调用都将在自定义执行器上执行，但事实并非如此。在调试期间，我看到逻辑在不同的线程池上执行:aws-java-sdk-NettyEventLoop-0-X、ForkJoinPool.commonPool-worker-X。AWS SNS 异步调用示例:

snsAsyncClient.publish(publishRequest)
        .thenApplyAsync(..)
        .whenCompleteAsync(..);

如何设置自定义线程池，以便异步非阻塞 IO 操作将在特定线程池上执行？

AWS SDK版本:software.amazon.awssdk sdk-core 2.7.3

Answer 1

异步配置SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR的思想是提供线程池执行器，它将用于CompletableFuture上的后续调用，例如thenApply(..) 和 whenComplete(..)。由于 CompletableFuture 实现，此执行程序不会应用于 thenApplyAsync 和 whenCompleteAsync 等方法(如果执行程序未传递到 thenApplyAsync >，默认情况下将使用ForkJoinPool.commonPool()，或者我们可以将自定义执行器作为第二个方法参数传递)。

snsAsyncClient.publish(publishRequest)
        .thenApply(..)
        .whenComplete(..);

thenApply 和 whenComplete 中的代码将在 FUTURE_COMPLETION_EXECUTOR 配置的执行器上进行处理。

为了设置异步客户端配置并提高性能，我们应该创建 NettyNioAsyncHttpClient.Builder 实例并使用方法 httpClientBuilder 将其传递给异步客户端

var asyncHttpClientBuilder = NettyNioAsyncHttpClient.builder()
        .maxConcurrency(100)
        .connectionTimeout(Duration.ofSeconds(20))
        .connectionAcquisitionTimeout(Duration.ofSeconds(20));
SnsAsyncClient.builder()
        .httpClientBuilder(asyncHttpClientBuilder)
        .build();