在Play!与Akka中动态设置CompletionStages的线程池大小

首先，你的代码里有个关键问题：你用了Java原生的CompletableFuture.thenComposeAsync，但没有把Akka的ExecutionContext作为第二个参数传入，所以默认会使用JDK的ForkJoinPool.commonPool()，完全没用到你配置的Akka调度器——这就是为什么你的线程池切换没生效。

第一步：让CompletionStage正确使用Akka的自定义调度器 #

Akka的ExecutionContext实现了Java的Executor接口，所以可以直接传给thenComposeAsync的第二个参数，这样就能让异步任务跑在你配置的线程池里。修改后的代码示例：

// 先获取你配置的自定义调度器
final ExecutionContext customDispatcher = system.dispatchers().lookup("my-dispatcher");

// 在thenComposeAsync中指定这个调度器
CompletableFuture.completedFuture(firstResult)
    .thenComposeAsync(firstResult -> {
        // 这里的逻辑会跑在my-dispatcher线程池里
        return doStuffThatCallsExternalService(firstResult);
    }, customDispatcher); // 关键：传入Akka的ExecutionContext作为Executor

第二步：动态调整Akka线程池大小的方法 #

Akka支持运行时动态调整调度器的线程池参数，不需要重启应用，具体分两种常见线程池类型处理：

1. 针对fork-join-executor（Akka默认线程池类型） #

如果你的my-dispatcher配置是这样的：

my-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"
  fork-join-executor {
    parallelism-min = 2
    parallelism-max = 8
    parallelism-factor = 3.0
  }
}

可以通过修改配置并重新加载调度器来动态调整：

// 获取当前配置并修改并行数参数
Config updatedConfig = system.settings().config()
    .withValue("my-dispatcher.fork-join-executor.parallelism-min", ConfigValueFactory.fromAnyRef(3))
    .withValue("my-dispatcher.fork-join-executor.parallelism-max", ConfigValueFactory.fromAnyRef(10));

// 更新ActorSystem的配置
system.settings().update(updatedConfig);

// 重新查找调度器，让新配置生效
ExecutionContext updatedDispatcher = system.dispatchers().lookup("my-dispatcher");

Akka的fork-join调度器会自动根据新参数调整线程池大小，无需额外操作。

2. 针对thread-pool-executor #

如果你的调度器用的是固定线程池类型：

my-dispatcher {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    core-pool-size-min = 2
    core-pool-size-max = 5
    max-pool-size = 10
  }
}

动态调整的逻辑类似，修改对应配置参数后重新加载调度器即可：

Config updatedConfig = system.settings().config()
    .withValue("my-dispatcher.thread-pool-executor.core-pool-size-max", ConfigValueFactory.fromAnyRef(6))
    .withValue("my-dispatcher.thread-pool-executor.max-pool-size", ConfigValueFactory.fromAnyRef(12));

system.settings().update(updatedConfig);
ExecutionContext updatedDispatcher = system.dispatchers().lookup("my-dispatcher");

额外建议：更灵活的并发控制方式 #

如果你的核心需求是控制外部服务的并行请求数，除了调整线程池大小，还可以用Semaphore做更精细的控制——比如不管线程池多大，只允许最多N个请求同时发出去：

// 初始化一个允许5个并行请求的信号量
Semaphore semaphore = new Semaphore(5);

CompletableFuture.completedFuture(firstResult)
    .thenComposeAsync(firstResult -> {
        semaphore.acquireUninterruptibly(); // 获取许可
        try {
            return doStuffThatCallsExternalService(firstResult)
                .whenComplete((result, err) -> semaphore.release()); // 完成后释放许可
        } catch (Exception e) {
            semaphore.release();
            throw e;
        }
    }, customDispatcher);

这种方式不依赖线程池配置，能更精准地控制对外部服务的请求并发量，适合不想调整全局线程池的场景。

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