使用 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

我经常遇到 OutOfMemory 错误，处理速度很慢。我认为这是因为每个异步方法都在一个单独的线程中执行，该线程阻塞了 I/O？对于OOME，我在第二点解释。关于缓慢，它确实与请求/响应处理中执行的 I/O 有关。问题来自并行有效运行的线程数量。使用您的实际配置，池的最大数量永远不会达到（我在下面解释原因）。假设corePoolSize==10在你的情况下。这意味着10个线程并行运行。假设每个线程运行大约 3 秒来测试站点。这意味着您在大约 0.3 秒内测试一个站点。测试 1000 个站点需要 300 秒。它足够慢，时间的重要部分是等待时间：从当前测试的站点发送/接收请求/响应的 I/O。为了提高整体速度，您最初可能应该并行运行比您的核心容量多得多的线程。通过这种方式，I/O 等待时间将不再是一个问题，因为线程之间的调度会很频繁，因此当这些线程暂停时，您将有一些对线程没有价值的 I/O 处理。它应该可以处理 OOME 问题，并且可能会大大提高执行时间，但不能保证您得到的时间很短。要实现它，您可能应该更精细地处理多线程逻辑并依赖具有非阻塞 IO 的 API/库。应该有帮助的官方文档的一些信息。这部分解释了提交任务时的整体逻辑（重点是我的）：线程池的配置也要结合executor的队列容量来考虑。有关池大小和队列容量之间关系的完整描述，请参阅 ThreadPoolExecutor 的文档。主要思想是，当提交任务时，如果当前活动线程数小于核心大小，则执行程序首先尝试使用空闲线程。如果已达到核心大小，则将任务添加到队列中，只要其容量尚未达到。只有这样，如果队列的容量已经达到，执行程序才会创建一个超出核心大小的新线程。如果也达到了最大大小，则执行者拒绝该任务。这解释了对队列大小的影响（重点仍然是我的）：默认情况下，队列是无界的，但这很少是所需的配置，因为如果在所有池线程都忙的情况下将足够多的任务添加到该队列中，它可能会导致 OutOfMemoryErrors。此外，如果队列是无界的，则最大大小根本没有影响。由于执行程序总是在创建超出核心大小的新线程之前尝试队列，因此队列必须具有有限的容量才能使线程池超过核心大小（这就是为什么固定大小的池是使用时唯一合理的情况一个无界队列）。长话短说：您没有设置默认为无界（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;}或者作为替代使用具有相同初始和最大池大小值的固定大小池：执行程序的线程池可以有不同的核心值和最大大小，而不是只有一个大小。如果您提供单个值，则执行程序具有固定大小的线程池（核心和最大大小相同）。public Executor proxyTaskExecutor() {    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();    executor.setCorePoolSize(100);    executor.setMaxPoolSize(100);    executor.setDaemon(true);    return executor;}另请注意，在没有暂停的情况下调用 1000 次异步服务在内存方面似乎是有害的，因为它无法直接处理它们。您可能应该通过在它们之间执行 thread.sleep() 将这些调用分成更小的部分（2、3 或更多）。