如何将 AWS SQS 入站适配器连接到 AMQP 支持的消息通道？答案

【问题标题】：How do I connect an AWS SQS Inbound Adapter to an AMQP Backed Message Channel?如何将 AWS SQS 入站适配器连接到 AMQP 支持的消息通道？
【发布时间】：2019-06-06 22:05:12
【问题描述】：

我已通过 Java 使用直接通道成功地为 AWS SQS 配置了入站通道适配器。该项目使用 JDBC 和 RabbitMQ 与 SQS 的组合作为 3 个独立队列的入站流。我需要持久消息，所以我试图弄清楚如何利用 RabbitMQ。我对如何引用 AMQP 通道感到困惑。如何从 setOutPutChannel 引用 AMQP 消息通道？我的目标是仅在消息成功发布到 RabbitMQ 持久队列时才将消息从 SQS 中删除。

@Bean
public MessageProducer getSQSChannel() {
    SqsMessageDrivenChannelAdapter adapter = new SqsMessageDrivenChannelAdapter(this.amazonSqs, MY_SQS_QUEUE);
    adapter.setOutputChannel(????);  
    return adapter;
}

@Bean
public AmqpChannelFactoryBean messageDriven(ConnectionFactory connectionFactory) {
    AmqpChannelFactoryBean factoryBean = new AmqpChannelFactoryBean(true);
    factoryBean.setConnectionFactory(connectionFactory);
    factoryBean.setQueueName("bar");
    factoryBean.setPubSub(false);
    return factoryBean;
}

【问题讨论】：

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

【解决方案1】：

使用

adapter.setOutputChannelName("messageDriven");

工厂 bean 创建的通道将在运行时根据其名称解析。

编辑

或者，正如阿尔乔姆所说...

MessageProducer getSQSChannel(MessageChannel messageDriven) { 
    ... 
}

【讨论】：

你也可以做方法参数注入MessageProducer getSQSChannel(MessageChannel messageDriven) { ... }
谢谢！以确认我的理解。如果我有一个集成流，例如轮询 SQS 消息、转换然后将 DirectChannel 作为传入通道发送到 RabbitMQ，那么在某些情况下可能会丢失消息？ SQS>>>InboundChannelAdapter>>>DirectChannel>>ObJTransformer>>PayloadEnricher>>RabbitOutboundAdapter 在我测试杀Rabbit之类的操作时，将sqs适配器删除策略设置为onSuccess，消息似乎没有被正确删除（但增加了Read Count ）。那么在这种情况下使用直接通道是否安全？
您不应该在 cmets 中向现有答案提出不相关的问题；它无助于社区找到问题/答案。 DirectChannel 是“安全的”，因为发布将发生在调用线程上。我不熟悉 SQS 扩展，所以我无法评论 onSuccess 策略何时适用以及它在这种情况下是否有效（我的猜测是它可能没问题，因为它是一个消息驱动的适配器，所以流程继续运行适配器螺纹）。
也就是说，RabbitMQ 发布无论如何都是异步的，因此您需要使用 RabbitMQ 事务来确保代理接受消息；事务是 RabbitMQ 发布的重大开销；尤其是只发布一条消息时。
谢谢加里。我会记住这一点。公平地说，我在问题的目标中陈述了不丢失信息的愿望，所以我认为这与问题有关。但我认识到可能有一个关于该特定主题的单独问题主题行，所以我会记住这一点。再次感谢！