Spring Integration 将入站通道适配器馈送到 DB 和 Kafka

Question

我正在重构我的一些 Spring Integration RSS 提要代码，它使用提要入站通道适配器到微服务。我希望将提要内部存储在 mongodb 数据库中（以防出现故障、审计等），并将提要（以 JSON 格式）写入 kafka 主题以进行后续处理。

如何使用 Spring Integration 做到这一点？我需要一个带有两个处理程序的发布/订阅队列吗？

任何使用 Java DSL 的示例代码都会很有帮助。

Answer 1

不确定您是否需要 pub/sub 来处理这种用例，但绝对可以在 Feed Inbound Channel Adapter 和 Kafka Outbound Channel Adapter 之间使用 QueueChannel。 QueueChannel 确实可以与 MongoDbChannelMessageStore 一起提供，以便在发生故障时保持持久性。

Java DSL 示例如下：

    @Bean
    public IntegrationFlow feedFlow(MongoDbChannelMessageStore messageStore) {
        return IntegrationFlows
                .from(Feed.inboundAdapter(...),
                        e -> e.poller(p -> p.fixedDelay(1000)))
                .channel(c -> c.queue(messageStore, "entries"))
                .handle(Kafka.outboundChannelAdapter(...), 
                        e -> e.poller(p -> p.fixedDelay(1000)))
                .get();
    }

https://docs.spring.io/spring-integration/docs/5.0.3.RELEASE/reference/html/feed.html#feed-java-configuration

https://docs.spring.io/spring-integration/docs/5.0.3.RELEASE/reference/html/mongodb.html#mongodb-priority-channel-message-store

https://docs.spring.io/spring-kafka/docs/2.1.4.RELEASE/reference/html/_spring_integration.html#si-outbound