RxJava：网络调用的线程池

Question

我正在尝试创建一种机制来限制并发网络请求的数量。我的想法是我想要一个固定的线程池，比如 20 个线程，并使用该池最多只允许 20 个传出 HTTP 请求。

我一直在尝试做的是：

public class HttpClient {
  private final Scheduler scheduler;

  public HttpClient(int maxRequests) {
    this.scheduler = Schedulers.from(Executors.newFixedThreadPool(maxRequests));
  }

  public Single<...> request() {
    return this.httpRequest()
      .subscribeOn(this.scheduler);
  }

  // sends the http request and returns a response
  private Single<...> httpRequest() {
    return ...
  }
}

但这不起作用。我尝试将maxRequests 设置为 1，发送 5 个请求，然后在接收请求的服务器上设置一个断点，以便将第一个请求“卡”在那里，以查看其他 4等待一个可用的线程。但是所有 5 个请求都执行了，过了一会儿，我在所有 5 个请求上都得到了一个超时异常。

我也尝试过使用observeOn，但也没有用。

编辑：我还尝试使用以下代码实现信号量逻辑：

public HttpClient(int maxRequests) {
  this.concurrentRequestsSemaphore = new Semaphore(maxRequests, true);
}

public Single<...> request() {
  return Completable.fromAction(concurrentRequestsSemaphore::acquireUninterruptibly)
   .andThen(this.httpRequest())
   .doFinally(concurrentRequestsSemaphore::release);
}

Semaphore 是信号量的本地 Java 实现。这个机制按预期工作，如果maxRequests 是 2，我发送了 5 个请求，2 个会出去，另外 3 个会卡在 fromAction 内等待。但是这种方法带来了其他意想不到的行为，例如即使在 2 个请求收到响应之后，其他 3 个请求都没有被执行，因为 .doFinally(concurrentRequestsSemaphore::release) 从未被执行。我做了一些测试，它只在 X 请求收到响应后执行。 X 会是什么完全无法预测。因此，可能有 20 个许可的信号量，20 个请求会发出并返回响应，而没有其他请求会被执行，因为该信号量从未被任何请求释放。

Answer 1

您没有显示private Single<...> httpRequest() 的正文。我假设你在那里调用了一些异步方法。异步方法只占用线程来处理响应，当请求本身移动到服务器并返回时，没有使用任何线程。这就解释了为什么您会看到所有 5 个请求都到达了服务器。 通常，为了限制某种活动的数量，使用java.util.concurrent.Semaphores，但它们通过阻塞线程来限制活动。从逻辑上讲，由于您的程序是异步的，因此您需要使用异步信号量，但它是一种罕见的野兽。 所以你有以下选择：

• 根本不限制异步 http 请求的数量，因为它们无论如何都不会占用太多资源
• 启动一个特殊线程，从普通同步信号量中获取许可，然后启动异步http请求。当请求完全完成时释放信号量
• 使用固定线程池同步启动http请求
• 使用异步信号量。我知道的唯一实现在我的库中DF4J： AsyncSemaphore 是标准 Semaphore 的扩展，因此具有同步和异步接口，InpSignal 仅用于异步程序。 InpSignal 的使用示例是 AsyncServerSocketChannel.java，它用于限制 Echo Server 实现中打开的客户端连接数。