使用 Spring WebClient 的内存泄漏

Question

我想查看域名列表（数百万条记录），发送请求并接收响应，以确定它是否存在。

我选择了一种反应式方法，我希望它可以只用很少的线程为大量主机提供服务，但我注意到我的堆内存一直在增长，直到达到 OutOfMemory。

这是我的代码：

@Slf4j
@Component
@RequiredArgsConstructor

public static class DataLoader implements CommandLineRunner {

    private final ReactiveDomainNameRepository reactiveDomainNameRepository;

    @Override
    @SneakyThrows
    public void run(String... strings) {
        ReactorClientHttpConnector connector = getConnector(); // Trying to reuse connector instead of creating new each time

        reactiveDomainNameRepository.findAllByResourcesIsNull() // Flux<DomainEntity>. This basically streams data from MongoDB using reactive driver
                .publishOn(Schedulers.parallel())
                .flatMap(domain -> performRequest(connector, domain)) // If I remove this line everything starts working just fine
                .buffer(1000) // Little optimization. The problem with memory remains even if I don't use buffering.
                .flatMap(reactiveDomainNameRepository::saveAll)
                .subscribe();
    }

    private Mono<DomainEntity> performRequest(ReactorClientHttpConnector connector, DomainEntity domain) {
        return WebClient
                .builder()
                .clientConnector(connector)
                .baseUrl("http://" + domain.getHost())
                .build()
                .get()
                .exchange()
                .onErrorResume(error -> {
                    log.error("Error while requesting '{}': ", domain.getHost());

                    return Mono.empty();
                }) // Mono<ClientResponse>
                .flatMap(resp -> {

                    if (resp.statusCode() == OK) {
                        log.info("Host '{}' is ok", domain.getHost());
                    } else {
                        log.info("Host '{}' returned '{}' status code", domain.getHost(), resp.statusCode().value());
                    }

                    // Consuming response as described in Spring documentation. Tried also resp.toEntity(String.class) but got the same result
                    return resp.toEntity(Void.class)
                            .map(nothing -> domain);
                });
    }
}

这里是堆内存使用情况。不要注意 5:59 - 6:05 的时间段——那是应用程序停止处理数据的地方，因为我没有处理极端情况。通常，它只会不断增长，直到达到内存限制。

所以我基本上有两个问题：

1. 我的代码有什么问题？
2. 使用响应式方法向不同主机发出大量请求是个好主意吗？

Answer 1

只需使用retrieve() 而不是exchange()，您甚至不需要杂乱的错误处理。

我知道这是一个迟到的回复，但我前段时间遇到了完全相同的问题并且碰到了你的问题，我只是想把这个可能的解决方案留在这里。 :)

并回答您的问题：

1. 使用exchange()时，你负责处理连接和错误处理，不是很推荐，只有在你确实需要控制时才应该使用。

2. 好吧，您正在使用并行构建，所以为什么不呢。