关于实现单域名单并发请求的响应式网络爬虫的问题咨询

你的问题出在**flatMap默认的并发行为**上——虽然你用groupBy按域名分组，也加了delayElements，但flatMap默认会同时处理多个元素（默认并发数是256），这就导致同一域名下的请求还是会被并行执行，没法达到并发1的限制。

核心问题分析 #

delayElements(Duration.ofSeconds(1))只是给每个元素添加了1秒的延迟，但并没有限制下游操作的并发数。比如，分组后的流里有3个元素，它们会分别延迟1秒、2秒、3秒，但到了flatMap(this::downloadDocument)这一步，三个请求可能会同时被触发（只要前一个延迟到点），还是会出现同一域名多请求并发的情况。

解决方案：用concatMap或指定并发数的flatMap #

要实现单域名并发1，你需要让分组后的流顺序执行每个请求，同时保持1秒的间隔。有两种常用方式：

1. 使用concatMap（推荐）

concatMap会严格按顺序处理流中的元素，前一个元素的处理完成后才会开始下一个，天然保证并发数为1。修改你的代码如下：

Flux.generate(downloaderQueueConsumer)
    .doFirst(this::initializeProcessing)
    .flatMap(this::evaluateDocumentLocation)
    .groupBy(this::parseDocumentDomain, 100000)
    .flatMap(documentSourceItem -> 
        documentSourceItem
            .delayElements(Duration.ofSeconds(1))
            .doOnNext(this::incrementProcessedCount)
            .concatMap(this::downloadDocument) // 顺序执行，并发1
            .concatMap(this::archiveDocument)   // 同理，归档也顺序执行
            .doOnNext(this::incrementArchivedCount)
    )
    .doFinally(this::finishProcessing)
    .subscribe();

2. 在flatMap中指定并发数为1

如果你更倾向于用flatMap，可以通过第二个参数明确设置并发数为1，效果和concatMap一致：

Flux.generate(downloaderQueueConsumer)
    .doFirst(this::initializeProcessing)
    .flatMap(this::evaluateDocumentLocation)
    .groupBy(this::parseDocumentDomain, 100000)
    .flatMap(documentSourceItem -> 
        documentSourceItem
            .delayElements(Duration.ofSeconds(1))
            .doOnNext(this::incrementProcessedCount)
            .flatMap(this::downloadDocument, 1) // 指定并发数1
            .flatMap(this::archiveDocument, 1)   // 归档也指定并发1
            .doOnNext(this::incrementArchivedCount)
    )
    .doFinally(this::finishProcessing)
    .subscribe();

额外注意点 #

• delayElements的位置：确保它在concatMap/flatMap之前，这样每个请求之间会先等待1秒，再执行下载操作，严格保证间隔。
• 分组稳定性：groupBy的第二个参数（100000）是分组的最大数量，这个设置没问题，避免过多分组导致资源消耗。

这样修改后，每个域名下的请求会严格按照“等待1秒 → 执行下载 → 完成归档 → 等待1秒 → 下一个请求”的顺序执行，完美满足你“单域名并发1+请求间隔1秒”的需求。

内容的提问来源于stack exchange，提问作者Lakatos Gyula