java - 如何保持固定大小的 ListenableFuture 池？

Question

我正在读取一个大的 URL 文件并向服务发出请求。该请求由返回 ListenableFuture 的客户端执行。现在我想保留一个 ListenableFuture 池，例如最多同时执行 N 个 Future。

我看到的问题是，由于第三方库的原因，我无法控制执行ListenableFuture的ExecutorService。否则，我只需创建一个 FixedSizePool 并创建我自己的 Callable。

1) 一个简单的实现是生成 N 个 Future，然后使用 AllAsList ，这将满足固定大小标准，但使所有请求都等待最慢的请求。乱序处理没问题。

2) 一个稍微好一点的简单选择是使用第一个想法并将其与速率限制器结合起来，通过设置 N 和 rate 来限制金额并发请求的数量非常接近所需的池大小。但实际上我并不是在寻找一种方法来限制调用，例如使用RateLimiter。

3) 最后一种选择是生成 N 个 Future，并使用回调来生成新的 Future。这满足了固定大小的标准并最大限度地减少了空闲时间，但是我不知道如何检测我的程序的结束，即关闭文件。

4) 与 ListenableFuture 无关的方法是直接 .get() 结果，并通过创建一个简单的 来处理令人尴尬的并行任务>线程池。

为了知道作业队列为空，即关闭文件，我正在考虑使用 CountdownLatch。这应该适用于许多选项。

Answer 1

嗯。您对仅使用 java.util.concurrent.Semaphore 感觉如何？

Semaphore gate = new Semaphore(10);
Runnable release = gate::release; // java 8 syntax.
Iterator<URL> work = ...;
while(work.hasNext() && gate.acquire()) {
  ListenableFuture f = ThirdPartyLibrary.doWork(work.next());
  f.addListener( release, MoreExecutors.sameThreadExecutor() );
}

您可以通过使用 Futures.addCallback(ListenableFuture, FutureCallback) 添加其他监听器来对结果执行某些操作，只要您小心 release() 关于成功和错误。

它可能会起作用。