Reactor的onErrorContinue操作符能否让原序列继续执行？

onErrorContinue的定位 #

首先要明确：onErrorContinue 不属于常规的错误处理操作符，它的实现逻辑和 onErrorResume/onErrorReturn 这类捕获全局onError终止信号的操作符完全不同。
常规错误处理操作符的逻辑是：上游抛出全局onError信号后，替换整个上游终止的序列，所以原序列肯定不会继续执行。而onErrorContinue本质是向下游传递了一个错误处理上下文，只有主动实现了该上下文感知的上游操作符（比如Reactor内置的map、flatMap等），才会在单个元素处理出错时，不把错误升级为全局onError终止信号，而是跳过当前元素、把错误交给onErrorContinue的回调处理后继续执行后续元素。如果上游操作符不识别这个上下文，onErrorContinue就完全不生效，还是会触发全局序列终止。

通用的忽略错误继续执行方案 #

通用方案的核心逻辑是：把每个元素的处理逻辑封装为独立的内部流，所有错误在内部流中消化，永远不要让错误泄漏到外层的主序列。这种方案对所有操作符都生效，不需要上游支持任何特殊机制，也是官方推荐的替代onErrorContinue的标准方案。
示例逻辑如下：

Flux<SourceElement> source = ... // 任意源Flux
source
  // 根据顺序需求可以替换为concatMap/flatMapSequential
  .flatMap(element -> processElement(element)
    // 单个元素的处理错误在内部捕获
    .doOnError(e -> log.error("处理元素失败: {}", element, e))
    // 错误时返回空，相当于跳过当前元素
    .onErrorResume(e -> Mono.empty())
  )
  // 外层主序列永远不会收到错误信号，会持续运行
  .subscribe();

Kafka消费代码优化建议 #

你当前的代码思路是正确的，以下几个优化点可以进一步提升消费稳定性：

1. 删除多余的repeat(() -> true)
KafkaReceiver.create(receiverOptions).receive()本身是无限流，正常场景下不会发送onComplete信号，repeat的作用是上游完成后重新订阅，在这里没有实际作用，可以直接移除。
2. 优化重试策略
   当前的Retry.indefinitely()没有退避机制，如果遇到Kafka broker宕机、网络闪断等问题，会无限立即重试，容易打满本地CPU、触发broker限流。建议改为带退避的重试，同时增加重试日志方便排查：

.retryWhen(Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
  .maxBackoff(Duration.ofSeconds(10))
  .doBeforeRetry(retrySignal -> log.warn("Kafka消费流异常，第{}次重试", retrySignal.totalRetries() + 1, retrySignal.failure()))
)

3. 增加单条消息处理超时
   如果processRequest出现阻塞、死锁等问题，会导致整个消费流卡住。建议给处理逻辑加上超时时间，超时后主动报错、ack消息继续消费：

.flatMap(record -> processRequest(record.value())
  // 超时时间根据业务场景调整
  .timeout(Duration.ofSeconds(30))
  .doOnNext(e -> record.receiverOffset().acknowledge())
  .doOnError(e -> {
    log.error("处理消息失败: {}", record.value(), e);
    // 可选：如果业务允许，把出错的消息发送到死信队列，方便后续回溯
    // produceToDeadLetterTopic(record.value(), e);
    record.receiverOffset().acknowledge();
  })
  .onErrorResume(e -> Mono.empty())
)

4. 补充subscribe的兜底错误回调
   虽然有重试机制，但极端情况下如果出现无法重试的错误，默认无参的subscribe()会把错误抛到调用栈，可能导致线程退出。建议添加兜底的错误回调：

.subscribe(
  v -> {},
  e -> log.error("Kafka消费流出现不可恢复错误", e)
);

优化后的代码可以覆盖几乎所有异常场景，满足你持续消费Kafka的需求。

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