Spring Batch RemoteChunking 出现 JMS 超时错误

2026-10-24支付老李

结论

Worker 返回的 ChunkResponse 已经是 Remote Chunking 协议的最终回复，Master 不需要再回一条消息。

当前 Worker 使用 Jms.outboundGateway() 发送 ChunkResponse。Gateway 采用 JMS request-reply 语义：发送消息时会创建临时 JMSReplyTo 队列，然后等待对方回复。Master 只消费 ChunkResponse，不会响应 Worker，因此 5000ms 后会出现：

MessageTimeoutException: failed to receive JMS response within timeout of: 5000ms

解决方法是将 Worker 的 Jms.outboundGateway() 改为单向的 Jms.outboundAdapter()。

为什么 Job 已成功，Worker 仍然超时

当前消息链路如下：

Master
  -- ChunkRequest --> requests 队列
Worker
  -- ChunkResponse --> replies 队列
Master
  -- 不再回复 -->
Worker 的 JmsOutboundGateway 等待超时

JmsOutboundGateway 会把 Worker 发出的 ChunkResponse 当作新的 JMS 请求：

.handle(Jms
        .outboundGateway(connectionFactory())
        .requestDestination("replies"))

它会执行两个动作：

将 ChunkResponse 发送到 replies 队列。Master 因此能够收到响应并正常完成 Job。
为这条消息设置临时 JMSReplyTo，然后等待 Master 回复。

日志中的这一行就是 Gateway 创建并等待的临时回复队列：

ReplyTo: temp-queue://ID:...

但 Master 的 inboundFlow 只会将消息转发到本地 replies channel：

.from(Jms.messageDrivenChannelAdapter(connectionFactory)
        .destination("replies"))
.channel(replies())

它不是入站 Gateway，不会向 Worker 的临时队列发送消息。所以业务处理虽然已经成功，Gateway 仍在等待，最后报告超时。

正确的配置方式

在这种双队列配置中，Remote Chunking 包含两条单向消息流：

requests：Master → Worker
replies：Worker → Master

两边的 JMS 端点都应使用 Adapter，Worker 的回复端不应使用 Gateway。

Master：发送请求

Master 使用单向 outbound adapter，将 ChunkRequest 发送到 requests 队列：

@Bean
public DirectChannel requests() {
    return new DirectChannel();
}

@Bean
public IntegrationFlow jmsOutboundFlow(
        ActiveMQConnectionFactory connectionFactory) {

    return IntegrationFlows
            .from("requests")
            .handle(Jms.outboundAdapter(connectionFactory)
                    .destination("requests"))
            .get();
}

Master：接收回复

Master 通过 message-driven inbound adapter 消费 replies 队列，再将 ChunkResponse 放入本地 QueueChannel：

@Bean
public QueueChannel replies() {
    return new QueueChannel();
}

@Bean
public IntegrationFlow inboundFlow(
        ActiveMQConnectionFactory connectionFactory) {

    return IntegrationFlows
            .from(Jms.messageDrivenChannelAdapter(connectionFactory)
                    .destination("replies"))
            .channel(replies())
            .get();
}

ChunkMessageChannelItemWriter 通过 requests channel 发送 chunk，再从 replies channel 获取处理结果：

@Bean
@StepScope
public ChunkMessageChannelItemWriter<Record> itemWriter() {
    ChunkMessageChannelItemWriter<Record> writer =
            new ChunkMessageChannelItemWriter<>();

    writer.setMessagingOperations(messagingTemplate());
    writer.setReplyChannel(replies());

    return writer;
}

@Bean
public MessagingTemplate messagingTemplate() {
    MessagingTemplate template = new MessagingTemplate();
    template.setDefaultChannel(requests());
    template.setReceiveTimeout(2000);
    return template;
}

这里的 receiveTimeout 控制 Master 等待本地回复 channel 返回 ChunkResponse 的时间，与 Worker 日志中 JMS Gateway 的 5000ms 超时不是一回事。

Worker：接收请求

Worker 使用 message-driven inbound adapter 消费 requests 队列：

@Bean
public DirectChannel requests() {
    return new DirectChannel();
}

@Bean
public IntegrationFlow jmsIn(
        ActiveMQConnectionFactory connectionFactory) {

    return IntegrationFlows
            .from(Jms.messageDrivenChannelAdapter(connectionFactory)
                    .destination("requests"))
            .channel(requests())
            .get();
}

Worker：处理 chunk

ChunkProcessorChunkHandler 接收 ChunkRequest，处理完成后将 ChunkResponse 发送到本地 replies channel：

@Bean
@ServiceActivator(
        inputChannel = "requests",
        outputChannel = "replies",
        sendTimeout = "10000")
public ChunkProcessorChunkHandler<Record> chunkProcessorChunkHandler() {
    ChunkProcessorChunkHandler<Record> handler =
            new ChunkProcessorChunkHandler<>();

    handler.setChunkProcessor(
            new SimpleChunkProcessor<>(recordProcessor(), paymentWriter()));

    return handler;
}

具体的泛型声明可能因 Spring Batch 版本而异。如果当前 4.0.x API 不接受上述形式，可以保留原有的原始类型声明，消息流的配置原则不变。

Worker：发送最终回复

需要修改的是这一处：

@Bean
public DirectChannel replies() {
    return new DirectChannel();
}

@Bean
public IntegrationFlow outgoingReplies(
        ActiveMQConnectionFactory connectionFactory) {

    return IntegrationFlows
            .from("replies")
            .handle(Jms.outboundAdapter(connectionFactory)
                    .destination("replies"))
            .get();
}

也就是将原来的：

Jms.outboundGateway(connectionFactory())
        .requestDestination("replies")

替换为：

Jms.outboundAdapter(connectionFactory())
        .destination("replies")

outboundAdapter 只发送消息，不会创建临时回复队列，也不会等待另一条响应。

完整的消息方向

修改后的链路如下：

Master ChunkMessageChannelItemWriter
    ↓ 本地 requests channel
JMS outboundAdapter
    ↓ requests 队列
Worker messageDrivenChannelAdapter
    ↓ 本地 requests channel
ChunkProcessorChunkHandler
    ↓ 本地 replies channel
JMS outboundAdapter
    ↓ replies 队列
Master messageDrivenChannelAdapter
    ↓ 本地 replies QueueChannel
ChunkMessageChannelItemWriter 接收 ChunkResponse

注意事项

不要通过延长 Gateway 超时解决

增加 replyTimeout 或其他等待时间只能延后异常。Master 本来就不会再响应 Worker 发出的 ChunkResponse，所以无论等待多久，Gateway 最终都会超时。

异常可能导致消息被重新投递

日志中包含：

jms_redelivered=true
JMSXDeliveryCount=2

这通常表示监听器处理链抛出的 MessageTimeoutException 传播到了 JMS listener container，导致消息事务回滚，或者 broker 将本次消费判定为失败，于是重新投递原始 ChunkRequest。

这样可能导致 Worker 重复处理同一个 chunk。即使 Master 已成功收到第一次 ChunkResponse，重复执行仍可能造成数据库重复写入，或者重复调用外部服务。改用 outboundAdapter 后，正常发送不会再因为等待不存在的第二次响应而失败。

生产环境中仍应让 ItemWriter 的写入操作具备事务性或幂等性，以处理真正的 broker 重投递和进程故障。

队列名称相同，不代表同一组请求与回复会话

本地 channel 和 JMS destination 都使用了 requests、replies 这两个名称。这样的配置可以正常工作，但它们不是同一个对象：

DirectChannel、QueueChannel 是应用进程内的 Spring Integration channel。
requests、replies destination 是 ActiveMQ 中的 JMS queue。

配置时需要分清每一步是在本地 channel 之间传递，还是在 JMS queue 之间传递。

Gateway 也能实现 request-reply，但必须成对配置

理论上，可以采用 Master outbound gateway 与 Worker inbound gateway 组成 JMS request-reply 模式。但这是另一种拓扑，需要双方 Gateway 成对工作，并正确处理 JMSReplyTo 和关联 ID。当前 ChunkMessageChannelItemWriter 已通过独立的请求 channel 和回复 channel 工作，没有必要再叠加一层 JMS Gateway。

对于固定的 requests/replies 双队列模型，直接使用四个单向端点即可：

Master outboundAdapter
Worker messageDrivenChannelAdapter
Worker outboundAdapter
Master messageDrivenChannelAdapter

修复的核心是：Worker 发送 ChunkResponse 时使用 Jms.outboundAdapter()，不要使用 Jms.outboundGateway()。