为什么在 Spring Integration 中将 ID 和 TIMESTAMP 声明为瞬态标头？

Question

我正在尝试通过 Spring Integration 的 AMQP 入站/出站适配器发送/接收消息，而我面对的是this problem。

在找到 Gary 的回答 here 后，我开始调查我的应用是否正确设置了消息 ID。事实上，它是自动处理的here。

制作人看起来像this。

我故意发送错误消息，在消费者端我看到它的消息转换器失败here。

在那之后，消息被重新排队并再次无休止地重新处理。

在调试这个问题时，我注意到发送和接收消息的 ID 总是不同的。

进一步调试此事件后，得知Spring核心消息传递框架提供的标准ID字段标记为transient和transient headers are never getting mapped。

有问题吗？

• 为什么框架不能自动将MessageHeaders.ID映射到AmqpHeaders.MESSAGE_ID？
• 我是否应该将自定义标头字段指定为 ID 并实施 MessageKeyGenerator？
• 顺便说一句，我将如何使用新的 Java DSL 来做到这一点？

非常感谢！

干杯， 拉斯洛

---

更新：在无限循环中重新处理失败消息的原因是由其他原因引起的。

我已经设法解决了这个问题，方法是将 defaultRequeueRejected(false) 添加到 AMQP 入站适配器的侦听器容器配置中。

@Bean
public IntegrationFlow webhookInboundFlow(
    ConnectionFactory connectionFactory, ObjectMapper objectMapper,
    HeaderValueRouter webhookInboundRouter) {

  return IntegrationFlows
      .from(Amqp.inboundAdapter(connectionFactory, FORGETME_WEBHOOK_QUEUE_NAME)
          .configureContainer(s -> s.defaultRequeueRejected(false))
      )
      .log(INFO)
      .transform(new ObjectToJsonNodeTransformer(objectMapper))
      .route(webhookInboundRouter)
      .get();
}

Answer 1

为什么框架不能自动将 MessageHeaders.ID 映射到 AmqpHeaders.MESSAGE_ID？

这不是一个坏主意，至少作为一种选择。随时打开“改进”JIRA Issue。

Spring AMQP 的AbstractMessageConverter 有一个属性createMessageIds 可以生成消息id 头；这独立于 Spring 集成。

所以通常情况下，如果您需要消息 id 标头，您可以在出站适配器的 RabbitTemplate 上设置该属性。