为什么在 Spring 集成中将 ID 和 TIMESTAMP 声明为瞬态 headers？

Question

我正在尝试通过 Spring Integration 的 AMQP in/outbound 适配器 send/receive 消息，我正面临 .

找到 Gary 的答案后 , I started to investigate if my app sets a message ID correctly. In fact, it's taken care of automatically here。

制作人长得像this.

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

在那之后，消息不断地 re-queued 和 re-processed。

在调试这个问题的时候，发现发送和接收的消息ID总是不一样。

进一步调试此事件后，我了解到 Spring 的核心消息传递框架提供的标准 ID 字段被标记为瞬态和 transient headers are never getting mapped.

有问题吗？

为什么框架不能自动将 MessageHeaders.ID 映射到 AmqpHeaders.MESSAGE_ID？
我是否应该强制使用自定义 headers 字段作为 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

Why couldn't the framework automatically map MessageHeaders.ID to AmqpHeaders.MESSAGE_ID?

这不是一个坏主意，至少作为一种选择。随意打开一个'improvement' JIRA Issue.

Spring AMQP的AbstractMessageConverter有一个属性createMessageIds生成消息idheader；这与 Spring 集成无关。

所以通常情况下，如果您想要消息 ID header，您可以在出站适配器的 RabbitTemplate.

上设置属性

为什么在 Spring 集成中将 ID 和 TIMESTAMP 声明为瞬态 headers？

Why are ID and TIMESTAMP declared as transient headers in Spring Integration?

spring

spring-integration

spring-messaging

spring-integration-amqp

spring-integration-dsl