Spring Cloud DataFlow Rabbit Source：如何截取和丰富Source中的消息

Question

我已经通过一个典型的简单流程成功评估了 Spring Cloud DataFlow：source |处理器 |下沉。

对于部署，将有多个源输入此管道，我可以使用数据流标签来完成。一切顺利。

每个源都是一个不同的 rabbitmq 实例，因为处理器需要知道消息来自哪里（因为它必须回调源系统以获取更多信息），所以我想到的策略是丰富每个带有源系统标头详细信息的消息，然后透明地传递给处理器。

现在，我精通 Spring、Spring Boot 和 Spring Integration，但我不知道如何丰富数据流 source 组件中的每条消息。

源组件绑定到 org.springframework.cloud.stream.app.rabbit.source.RabbitSourceConfiguration。源使用默认的 Source.OUTPUT 通道。如何获取源中的每条消息以丰富它？

我的处理器组件使用一些 Spring Integration DSL 来完成它需要做的一些事情，但是这个处理器组件根据定义同时具有 INPUT 和 OUTPUT 通道。 RabbitSourceConfiguration 源并非如此。

那么，这样可以吗？

Answer 1

我认为您需要在 RabbitSourceConfiguration 中的 MessageListenerContainer 上自定义 MessageListener。

在RabbitSourceConfiguration 中，您可以在MessageListenerContainer 上设置自定义ChannelAwareMessageListener（您也可以从MessagingMessageListenerAdapter 扩展）来执行您想要做的事情。

Answer 2

最终起作用的是将 org.springframework.cloud.stream.app.rabbit.source.RabbitSourceConfiguration 子类化为：

• 覆盖 public SimpleMessageListenerContainer container() 以便我可以在调用 super.container() 之前插入自定义运行状况检查。我的业务逻辑通过消息来自何处的详细信息来丰富每条消息（参见下一个项目符号）（注意，这是消息的发布者，而不是兔子队列）。需要进行健康检查来验证额外的丰富信息（通过配置提供），以确保消息不会从队列中被消耗并使用错误的信息进行丰富。如果验证失败，则源组件无法启动，因此不会消耗任何消息。

• 覆盖 AmqpInboundChannelAdapter bean 的创建，以便可以在适配器上设置 DefaultAmqpHeaderMapper 的自定义子类。此自定义映射器在公共 Map toHeadersFromRequest(final MessageProperties source) 中添加丰富的标头。

对我来说，流/数据流无法拦截和修改 Source 组件中的消息是有问题的。我真的不应该像以前那样摆弄底层消息代理 API。我应该能够做到这一点，例如弹簧集成。事实上，我可以注册一个全局消息拦截器，但我不能更改消息的标题。

这个能力会出现在我的 WIBNI（如果不是很好）列表中。也许我会提出这个要求。