java - 使用 Spring Boot 在 Java 中发送异步 HTTP 请求

Question

我正在开发一个需要连续测试 1000 个代理服务器的应用程序。该应用程序基于 Spring Boot。

我目前使用的方法是 @Async 装饰方法，它采用代理服务器并返回结果。

我经常遇到 OutOfMemory 错误并且处理速度非常慢。我认为这是因为每个异步方法都在一个单独的线程中执行，该线程会阻塞 I/O？

在我读到的有关 Java 异步的所有内容中，人们都将线程中的并行执行与非阻塞 IO 混合在一起。在 Python 世界中，有一个在单线程中执行 I/O 请求的异步库。当一个方法等待服务器响应时，它开始执行其他方法。

我认为在我的情况下，我需要这样的东西，因为 Spring 的 @Async 不适合我。有人可以帮助消除我的困惑并建议我应该如何应对这个挑战吗？

我想同时检查 100 个代理而不增加过多的负载。我已阅读有关 Apache Async HTTP Client 的内容，但我不知道它是否合适？

这是我正在使用的线程池配置:

    public Executor proxyTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors() * 2 - 1);
        executor.setMaxPoolSize(100);
        executor.setDaemon(true);
        return executor;
    }

Answer 1

I am often getting OutOfMemory error and the processing is very slow. I assume that is because each async method is executed in a separate thread which blocks on I/O?

对于OOME，我在第二点解释。
关于慢，确实和request/response处理中执行的I/O有关。
问题来自并行有效运行的线程数。
使用您的实际配置，永远不会达到 pool max 的数量(我在下面解释原因)。假设您的情况是 corePoolSize==10。这意味着 10 个线程并行运行。假设每个线程运行大约 3 秒来测试站点。
这意味着您可以在大约 0.3 秒内测试一个站点。测试 1000 个站点需要 300 秒。
它足够慢，并且时间的重要部分是等待时间:从当前测试的站点发送/接收请求/响应的 I/O。
为了提高整体速度，您最初应该并行运行比核心容量更多的线程。通过这种方式，I/O 等待时间将不再是一个问题，因为线程之间的调度会很频繁，因此当线程暂停时，您会有一些对线程没有值(value)的 I/O 处理。

---

它应该可以解决 OOME 问题并可能大大缩短执行时间，但不能保证您会得到很短的时间。
要实现它，您可能应该更精细地处理多线程逻辑并依赖具有非阻塞 IO 的 API/库。

the official documentation的一些信息|这应该会有帮助。
这部分解释了提交任务时的整体逻辑(强调是我的):

The configuration of the thread pool should also be considered in light of the executor’s queue capacity. For the full description of the relationship between pool size and queue capacity, see the documentation for ThreadPoolExecutor. The main idea is that, when a task is submitted, the executor first tries to use a free thread if the number of active threads is currently less than the core size. If the core size has been reached, the task is added to the queue, as long as its capacity has not yet been reached. Only then, if the queue’s capacity has been reached, does the executor create a new thread beyond the core size. If the max size has also been reached, then the executor rejects the task.

这解释了对队列大小的影响(重点仍然是我的):

By default, the queue is unbounded, but this is rarely the desired configuration, because it can lead to OutOfMemoryErrors if enough tasks are added to that queue while all pool threads are busy. Furthermore, if the queue is unbounded, the max size has no effect at all. Since the executor always tries the queue before creating a new thread beyond the core size, a queue must have a finite capacity for the thread pool to grow beyond the core size (this is why a fixed-size pool is the only sensible case when using an unbounded queue).

长话短说:您没有设置默认情况下无限制的队列大小 (Integer.MAX_VALUE)。因此，您用数百个任务填充队列，这些任务只会在很久以后弹出。这些任务使用大量内存，而 OOME 上升。

此外，如文档中所述，此设置对无界队列无能为力，因为只有当队列已满时才会创建新线程:

executor.setMaxPoolSize(100);

同时设置相关值的信息更有意义:

public Executor proxyTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(Runtime.getRuntime().availableProcessors() * 2 - 1);
    executor.setMaxPoolSize(100);
    executor.setQueueCapacity(100); 
    executor.setDaemon(true);
    return executor;
}

或者作为替代使用 fixed-size pool初始和最大池大小的值相同:

Rather than only a single size, an executor’s thread pool can have different values for the core and the max size. If you provide a single value, the executor has a fixed-size thread pool (the core and max sizes are the same).

public Executor proxyTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(100);
    executor.setMaxPoolSize(100);
    executor.setDaemon(true);
    return executor;
}

另请注意，不暂停地调用异步服务 1000 次似乎对内存有害，因为它无法直接处理它们。您可能应该通过在它们之间执行 thread.sleep() 将这些调用分成更小的部分(2、3 或更多)。