Spring Boot、Spring 集成 RabbitMQ 和 AWS Kinesis 集成答案

【问题标题】：Spring Boot , Spring Integration RabbitMQ & AWS Kinesis IntegrationSpring Boot、Spring 集成 RabbitMQ 和 AWS Kinesis 集成
【发布时间】：2021-09-15 22:15:08
【问题描述】：

我需要使用来自 rabbitMQ 的消息，进行一些处理，最后将消息发布到 Kinesis Data 流。我们已经在使用 Spring Boot、Spring Integration Core 和 Spring Integration AQMP 5.5.1 集成流来使用来自 RabbitMQ 的消息。我们的任何项目都没有使用 Spring Cloud Stream。

对于将消息发布到 Kinesis 数据流的用例，您建议使用什么 spring 库？在浏览了 Spring 文档之后，我看到了几个可用的选项。能否请您告知哪个是最好的追求？

spring-cloud-stream-binder-aws-kinesis
spring-integration-aws

【问题讨论】：

标签： spring-integration spring-cloud-stream spring-amqp spring-integration-aws

【解决方案1】：

只要您不关心 Spring Cloud Stream，就不应将 AWS Kinesis Binder for Spring Cloud Stream 依赖项带入您的项目。仅仅因为在您的项目中没有 Spring Cloud Stream 功能，它就无法工作。

由于您的应用程序确实是 Spring Integration 应用程序，因此您肯定需要使用 Spring Integration for AWS 依赖项。它带有一个KinesisMessageHandler 实现，用于将记录生成到 Kinesis 流中。

有关更多信息，请参阅其文档：https://github.com/spring-projects/spring-integration-aws#outbound-channel-adapter-3。这样的处理程序应声明为 bean，并且可以在 IntegrationFlow 定义的 .handle() 端点中使用。请参阅有关现有处理程序和错过的 Java DSL 工厂的文档：https://docs.spring.io/spring-integration/docs/current/reference/html/dsl.html#java-dsl-protocol-adapters

更新

如何在 Java DSL 中使用 Kinesis MH：

@Bean
public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis) {
    KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis);
    return kinesisMessageHandler;
}

@Bean
IntegrationFlow someFlow(MessageHandler kinesisMessageHandler) {
    return f -> f
              .handle(kinesisMessageHandler);
}

【讨论】：

感谢您的反馈。这很有帮助。我将根据我们的需要评估 spring-integration-aws。
是否有示例可用于参考使用 Kinesis 出站通道适配器的 IntegrationFlows DSL？
据我所知没有，但为什么呢？在我的回答中查看更新：拥有一个 bean 并将其注入另一个非常简单。