Spring Boot集成RabbitMQ高负载下RequestProcessor停止处理请求问题

代码实现 #

FooService.kt #

private val requestProcessor = Sinks.many().multicast().onBackpressureBuffer<FooRequest>(100)

val emitFailureHandler = EmitFailureHandler { _: SignalType?, emitResult: EmitResult ->
    (emitResult == EmitResult.FAIL_NON_SERIALIZED || emitResult == EmitResult.FAIL_OVERFLOW)
}

init {
    requestProcessor.asFlux()
      .flatMap { request ->
        try {
          logger.debug("handleByRabbitMq : {}", request)
          footGateway.handleByRabbitMq(request)
        } catch (e: Exception) {
          logger.error("unknown error when processing foo request", e)
          Mono.empty()
        }
      }
      .subscribe()
}

fun handleRequest(fooRequest: FooRequest): Mono<FooResponse> {
    val (_, id) = fooRequest
    val responseSink = Sinks.one<FooResponse>()
    responseListeners[id] = responseSink
    requestProcessor.emitNext(fooRequest, emitFailureHandler)
    return responseSink.asMono()
      .timeout(Duration.ofSeconds(10))
      .onErrorResume { Mono.just(FooResponse(REQUEST_SUBMITTED)) }
      .doOnTerminate { responseListeners.remove(id) }
}

FooController.kt #

@GetMapping("/foo/request")
@Timed(value = "foo.request")
fun fooRequest(fooRequest: FooRequest): Mono<String> {
    return performAction(fooRequest)
      .flatMap { fooResponse ->
        Mono.fromCallable {
          "FooResponse: $fooResponse"
        }
      }
      .doOnNext { logger.debug("response {}", it) }
}

fun performAction(fooRequest: FooRequest): Mono<FooResponse> {
    return Mono.just(fooRequest)
      .flatMap { request ->
        fooService.handleRequest(request)
      }
      .doOnNext { response ->
        //some operations here
        logger.debug(response)
      }.subscribeOn(Schedulers.boundedElastic())
}

问题现象 #

初始状态下请求可正常发送处理，但高负载或用k6压测时，requestProcessor停止发送消息：调用requestProcessor.emitNext(fooRequest, emitFailureHandler)后，init块中的消息处理逻辑不再执行。

最初会抛出FAIL_NON_SERIALIZED和FAIL_OVERFLOW错误，添加自定义emitFailureHandler后问题仅延迟数小时出现：

1. 应用仍能响应，但请求成功率降至96%；
2. 再次压测成功率跌至88%，最终抛出503 HTTP响应，不再接受请求。

排查发现 #

移除init块中RabbitMQ调用fooGateway.handleByRabbitMq(request)后，应用恢复正常响应，但未找到该调用的错误日志。

原因分析 #

1. RabbitMQ调用背压处理缺失：fooGateway.handleByRabbitMq(request)可能是阻塞调用，或返回的Mono未正确处理背压，导致flatMap并发队列堆积，耗尽资源后引发requestProcessor缓冲区溢出，最终流停滞。
2. 默认并发度不匹配：flatMap默认并发度为256，当RabbitMQ处理速度跟不上请求流入速度时，大量未完成的Mono堆积，触发背压机制，导致requestProcessor无法接收新消息。
3. 错误处理不完整：try-catch仅捕获同步异常，若handleByRabbitMq返回的Mono出现异步错误（如RabbitMQ连接超时、信道关闭），会被静默处理为Mono.empty()，无法及时发现问题，甚至导致流中断。
4. 溢出重试逻辑不合理：emitFailureHandler对FAIL_OVERFLOW直接重试，会加剧负载，导致缓冲区持续溢出，最终流停止处理。

解决方案 #

1. 控制RabbitMQ处理并发度 #

根据RabbitMQ集群能力调整flatMap并发度，避免过度堆积：

requestProcessor.asFlux()
  .flatMap({ request ->
    try {
      logger.debug("handleByRabbitMq : {}", request)
      fooGateway.handleByRabbitMq(request)
    } catch (e: Exception) {
      logger.error("unknown error when processing foo request", e)
      Mono.empty()
    }
  }, concurrency = 10) // 按实际处理能力设置合理值
  .subscribe()

2. 完善异步错误监控 #

不再静默处理异步错误，记录详细日志便于排查：

.flatMap({ request ->
  logger.debug("handleByRabbitMq : {}", request)
  fooGateway.handleByRabbitMq(request)
    .doOnError { e ->
      logger.error("RabbitMQ processing failed for request: {}", request, e)
    }
}, concurrency = 10)
.onErrorContinue { e, req ->
  logger.error("Failed to process request: {}", req, e)
}

3. 优化Sinks溢出策略 #

调整缓冲区大小，改用溢出丢弃策略并处理降级响应，避免无限制重试：

// 调整缓冲区大小，溢出时丢弃最旧消息并通知请求方
private val requestProcessor = Sinks.many().multicast().onBackpressureBuffer<FooRequest>(
  200, // 根据负载调整缓冲区容量
  false,
  { droppedRequest ->
    logger.warn("Request buffer overflow, dropped request: {}", droppedRequest)
    // 给对应请求返回降级响应
    val responseSink = responseListeners.remove(droppedRequest.id)
    responseSink?.tryEmitValue(FooResponse(REQUEST_SUBMITTED))
  }
)

// 调整失败处理器，仅在并发冲突场景重试
val emitFailureHandler = EmitFailureHandler { _, emitResult ->
  when (emitResult) {
    EmitResult.FAIL_NON_SERIALIZED -> true // 并发写入冲突可重试
    EmitResult.FAIL_OVERFLOW -> {
      logger.warn("Sink overflow when emitting request")
      false // 溢出不再重试，避免负载加剧
    }
    else -> false
  }
}

4. 确保RabbitMQ调用异步化 #

若handleByRabbitMq是阻塞调用，包装到弹性线程池执行：

fooGateway.handleByRabbitMq(request)
  .subscribeOn(Schedulers.boundedElastic())

5. 添加监控指标 #

监控缓冲区使用、处理耗时、错误率等指标，提前发现异常：

// 监控缓冲区大小
val bufferGauge = Gauge.builder("foo.request.buffer.size") {
  (requestProcessor as Sinks.ManySpec.MulticastSpec.OnBackpressureBufferSpec).currentBufferSize()
}.register(meterRegistry)

// 监控RabbitMQ处理耗时
fooGateway.handleByRabbitMq(request)
  .name("foo.rabbitmq.processing.time")
  .tag("request_id", request.id)
  .metrics()

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