JSON headerMode=raw 的 MessageHeaders

Question

我有一个 Spring Cloud Stream (SCS) Kafka 生产应用程序，配置了用于跟踪的 Sleuth。我正在尝试发布一个 spring-messaging 'GenericMessage'（即 MessageHeaders + payload），需要由以下两者使用：

• SCS 消费者（谁应该继续自动添加到 MessageHeaders 的跟踪）
• 非 SCS/Java 消费者应该跳过 MessageHeaders 并处理负载。

使用 headerMode=embeddedHeaders（默认）： MessageHeader 中的跟踪条目在消息发布之前由 EmbeddedHeaderUtils 添加到消息之前。我的非 java 消费者无法处理此问题，因为 EmbeddedHeaderUtils 不会序列化为纯 JSON，即

?\n invalid-json-headers { payload }

使用 headerMode=raw: MessageHeaders 根本没有发送，只有有效负载被序​​列化。 IE。 { payload }

我真的只想发布创建时的整个 GenericMessage，包括 Sleuth 添加的跟踪 + 跨度 ID，即：

{"headers": {"id": "x", "trace": "y", "span": "z"}, "payload": { ... }}

除了为 SCS 消费者发布到一个主题，而另一个仅使用负载发布之外，还有什么方法可以实现这一目标吗？

Answer 1

因此，将该消息作为出站消息的 payload 发布：

return MessageBuilder.withPayload(message).build();

这样您的消息将被序列化为 payload，您可以在适当的反序列化并将 payload 转换为 Message 后访问所需的 headers。