Spring Cloud Stream如何监听消费成败事件以统计Micrometer指标？

我正在构建Spring Boot应用，借助Spring Cloud Stream从消息中间件消费消息。请问Spring Cloud Stream框架是否存在监听机制，可通过回调获取消费操作成功或失败的事件？我需要结合Micrometer Counter对这些情况进行统计。

Spring Cloud Stream文档对错误处理机制有全面描述，但仅覆盖失败场景，我希望同时覆盖成功场景。

请问在消费Bean中增加成功处理消息的Counter计数是否合理（详见下方代码）？我对此存疑，因为消费Bean仅包含业务逻辑，担心Spring Cloud Stream内部逻辑可能在消费函数返回后出错，这类情况我也希望纳入统计。

@SpringBootApplication
class SampleApplication {

        val Counter success = Metrics.counter("success")
        val Counter failure = Metrics.counter("failure")

        fun main(args: Array<String>) {
          runApplication<SampleApplication>(*args)
        }

        @Bean
        fun consumer(): (Message<Payload>) -> Unit = { message ->
           System.out.println("Received: " + message.payload);
           doBusinessLogic()
           // NOT OK, will be incremented even in case of some error happened
           // in Spring Cloud Stream right after consumer function returns. 
           success.increment()
           return null
        }

        @Bean
        fun consumerErrorHandler(): (ErrorMessage) -> Unit = { error -> 
           // OK, will be incremented in any error case.
           failure.increment()
        }
        
}

1. 直接在消费函数中统计成功的问题 #

你担心的点完全正确：这种方式不可靠。消费函数返回仅代表业务逻辑执行完成，但Spring Cloud Stream在这之后还有一系列操作（比如消息偏移量提交、中间件消息确认等），如果这些环节出错，实际消费已经失败，但你的成功计数器已经完成累加，导致统计数据失真。

2. Spring Cloud Stream的全生命周期监听方案 #

Spring Cloud Stream底层依赖消息中间件的监听容器（如Kafka的ConcurrentMessageListenerContainer、RabbitMQ的SimpleMessageListenerContainer），这些容器提供了覆盖完整消费流程的回调接口，能精准捕获消费成功/失败的事件，包括框架层面的错误。

方案一：自定义监听容器回调（推荐）

通过ListenerContainerCustomizer为消费容器添加全局回调，覆盖业务逻辑执行后的框架操作环节：

@SpringBootApplication
class SampleApplication {

    @Bean
    fun successCounter(): Counter = Metrics.counter("message.consumption.success")

    @Bean
    fun failureCounter(): Counter = Metrics.counter("message.consumption.failure")

    fun main(args: Array<String>) {
        runApplication<SampleApplication>(*args)
    }

    @Bean
    fun consumer(): (Message<Payload>) -> Unit = { message ->
        println("Received: " + message.payload)
        doBusinessLogic()
    }

    @Bean
    fun consumerErrorHandler(): (ErrorMessage) -> Unit = { error ->
        // 处理业务逻辑抛出的错误
        failureCounter().increment()
    }

    @Bean
    fun listenerContainerCustomizer(
        successCounter: Counter,
        failureCounter: Counter
    ): ListenerContainerCustomizer<AbstractMessageListenerContainer<*, *>> {
        return ListenerContainerCustomizer { container, _, _ ->
            // 消息全流程处理完成后的回调（业务逻辑+框架操作）
            container.setAfterMessageProcessedCallback { _, exception ->
                if (exception == null) {
                    // 无异常表示消费全流程成功
                    successCounter.increment()
                } else {
                    // 框架层面的错误（如提交偏移量失败）
                    failureCounter.increment()
                }
            }
            // 容器级错误兜底处理
            container.errorHandler = ErrorHandler { throwable ->
                failureCounter.increment()
                // 复用默认错误处理逻辑
                DefaultErrorHandler().handleError(throwable)
            }
        }
    }

    private fun doBusinessLogic() {
        // 业务逻辑实现
    }
}

方案二：监听Spring事件

Spring Cloud Stream会在消费流程关键节点发布事件，你可以通过@EventListener监听这些事件来统计：

@Component
class ConsumptionMetricsListener(
    private val successCounter: Counter,
    private val failureCounter: Counter
) {

    // 监听消费成功事件（适用于函数式/StreamListener模型）
    @EventListener
    fun handleSuccess(event: StreamListenerMessageHandledEvent) {
        successCounter.increment()
    }

    // 监听消费失败事件
    @EventListener
    fun handleFailure(event: ListenerContainerConsumerFailedEvent) {
        failureCounter.increment()
    }
}

3. 关键说明 #

• 失败统计可以结合自定义错误处理器+容器回调/事件监听，覆盖业务逻辑错误和框架层面错误
• 如果使用Kafka，还可以通过ConsumerRecordInterceptor实现更细粒度的拦截，但容器回调方式更通用
• Spring Cloud Stream内置了部分Micrometer指标（如spring.cloud.stream.listener.messages.received），如果需求简单可以直接复用，但自定义统计更灵活

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