如何基于Spring Integration实现Pub/Sub批量消息监听？

问题背景 #

使用Google Pub/Sub订阅功能时，单条消息处理速度较慢，尝试改为批量处理时触发类型转换异常：

org.springframework.messaging.MessageHandlingException: error occurred during processing message in 'MethodInvokingMessageProcessor' [org.springframework.integration.handler.MethodInvokingMessageProcessor@7cd56906], failedMessage=GenericMessage [...
...
Caused by: java.lang.ClassCastException: class java.lang.Byte cannot be cast to class org.springframework.messaging.Message (java.lang.Byte is in module java.base of loader 'bootstrap'; org.springframework.messaging.Message is in unnamed module of loader 'app')

原单条处理代码可正常运行，但修改为接收List<Message<ByteArray>>参数时出现上述错误。

问题根源 #

默认情况下PubSubMessageSource以单条方式推送消息到通道，直接将处理器参数改为List会导致Spring Integration尝试拆分单条消息的ByteArray payload为单个Byte，再强制转换为Message，从而引发类型错误。要实现批量处理，需先配置消息源支持批量拉取，并调整处理器的参数类型。

解决方案步骤 #

1. 配置PubSubMessageSource开启批量拉取 #

修改入站通道适配器的配置，启用批量拉取并设置相关参数：

@Bean
fun inputMessageChannel(): MessageChannel {
    return PublishSubscribeChannel()
}

@Bean
@InboundChannelAdapter(channel = "inputMessageChannel")
fun inboundChannelAdapter(
    pubSubTemplate: PubSubTemplate
): MessageSource<Any> {
    val messageSource = PubSubMessageSource(pubSubTemplate, pubSubConfig.subscription)
    messageSource.setAckMode(AckMode.MANUAL)
    // 开启批量拉取模式
    messageSource.setBatchReceiveEnabled(true)
    // 设置每次拉取的最大消息数（可根据业务调整，上限1000）
    messageSource.setMaxMessagesPerPoll(10)
    // 设置拉取超时时间（可选，单位毫秒）
    messageSource.setReceiveTimeout(1000)
    return messageSource
}

2. 调整消息处理器接收批量消息 #

开启批量拉取后，PubSubMessageSource会发送封装了List<BasicAcknowledgeablePubsubMessage>的Message到通道，需调整处理器参数类型适配：

@ServiceActivator(inputChannel = "inputMessageChannel")
fun messageReceiver(
    message: Message<List<BasicAcknowledgeablePubsubMessage>>
) {
    try {
        val pubsubMessages = message.payload
        // 提取消息payload进行批量处理
        val messagePayloads = pubsubMessages.map { it.payload.toByteArray() }
        handleBatchMessages(messagePayloads)
        // 批量确认消息
        pubsubMessages.forEach { it.ack() }
    } catch (e: Exception) {
        log.error(e) { "Failed to handle messages batch." }
        // 可选：批量拒绝消息，根据业务需求调整失败处理逻辑
        message.payload.forEach { it.nack() }
    }
}

如果需要保持handleBatchMessages接收List<Message<ByteArray>>类型参数，可手动转换消息格式：

@ServiceActivator(inputChannel = "inputMessageChannel")
fun messageReceiver(
    message: Message<List<BasicAcknowledgeablePubsubMessage>>
) {
    try {
        val springMessages = message.payload.map { pubsubMsg ->
            MessageBuilder.withPayload(pubsubMsg.payload.toByteArray())
                .setHeader(GcpPubSubHeaders.ORIGINAL_MESSAGE, pubsubMsg)
                .build()
        }
        handleBatchMessages(springMessages)
        // 逐个确认消息
        springMessages.forEach {
            it.headers.get(GcpPubSubHeaders.ORIGINAL_MESSAGE, BasicAcknowledgeablePubsubMessage::class.java)?.ack()
        }
    } catch (e: Exception) {
        log.error(e) { "Failed to handle messages batch." }
    }
}

3. 关键注意事项 #

• setBatchReceiveEnabled(true)是开启批量处理的核心配置，必须设置。
• setMaxMessagesPerPoll的取值需结合业务吞吐量和Pub/Sub的限制（默认最大1000条/次）调整。
• 批量处理时建议统一处理ACK/NACK，若需单独处理每条消息的确认状态，需注意异常场景下的消息重复消费问题。

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