Spring Cloud Stream响应式管道异常致流崩溃：Dispatcher无订阅者

问题根源 #

你的消费者函数在WebClient调用失败时，会因为未处理的异常导致整个Flux流终止，订阅者取消订阅。后续消息发送到通道时，因没有订阅者触发Dispatcher has no subscribers错误，最终被移至死信队列。

具体原因：

• WebClient的retrieve()方法在收到4xx/5xx响应时，会抛出WebClientResponseException
• 代码未对该异常做任何处理，导致concatMap中的Mono出错后，错误向上传播终止整个Flux
• 直接调用无错误回调的subscribe()，异常触发ErrorCallbackNotImplemented，最终流崩溃，订阅者消失

解决方案 #

核心思路是在WebClient调用环节捕获异常，避免错误终止整个流，同时给订阅添加错误处理逻辑。

修改后的代码示例 #

@Bean
fun consumer() = Consumer<Flux<String>> { flux ->
    flux.concatMap { url ->
        WebClient.create()
            .get()
            .uri(url)
            .retrieve()
            // 可选：显式捕获错误状态码，转换为异常或直接返回处理结果
            .onStatus({ it.isError }) { response ->
                Mono.error(WebClientResponseException.create(
                    response.statusCode().value(),
                    response.statusCode().reasonPhrase(),
                    response.headers().asHttpHeaders(),
                    null,
                    null,
                    response.request()
                ))
            }
            .bodyToMono(String::class.java)
            // 捕获WebClient异常，记录日志后返回空Mono，让流继续处理下一个消息
            .onErrorResume { ex ->
                when (ex) {
                    is WebClientResponseException -> {
                        println("调用URL $url 失败，状态码: ${ex.statusCode}, 错误信息: ${ex.message}")
                    }
                    else -> {
                        println("调用URL $url 发生未知错误: ${ex.message}")
                    }
                }
                // 返回空Mono，确保流不会终止
                Mono.empty()
            }
    }
    // 订阅时添加错误回调，避免出现ErrorCallbackNotImplemented
    .subscribe(
        { result -> println("处理成功: $result") },
        { ex -> println("流全局错误: ${ex.message}") }
    )
}

关键修改说明 #

1. onErrorResume处理局部异常：在WebClient调用的Mono上添加onErrorResume，捕获异常并记录日志后返回Mono.empty()，这样concatMap会继续处理下一个消息，不会终止整个Flux流。
2. 带错误回调的subscribe()：使用subscribe的重载方法，添加错误消费逻辑，避免触发ErrorCallbackNotImplemented异常。
3. 可选的onStatus处理：显式捕获HTTP错误状态码，可根据业务需求选择转换为异常或直接返回自定义结果（比如默认值）。

额外注意事项 #

• 如果需要将失败的消息单独路由到DLQ，可在onErrorResume中发送消息到指定通道，替代返回Mono.empty()。
• 建议使用Spring Cloud Stream的函数式编程规范，避免直接手动调用subscribe()，让框架管理订阅生命周期，错误处理更可控。
• 原日志中的重试机制是针对消息消费的重试，但流崩溃后订阅者消失，重试无法生效，必须先保证流的稳定性。

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