如何在 spring-integration-aws 中使用 sqs-message-driven-channel-adapter

Question

编辑：这是显示我的日志的要点。似乎有 ReceiveMessage，然后 inputChannel 上有一个 preSend：

https://gist.github.com/louisalexander/04e7d95835521efdd15455c98075e2ea

抱歉这么密集，但我似乎无法弄清楚如何正确使用 sqs-message-driven-channel-adapter

在我的上下文文件中，我将其配置为：

<int-aws:sqs-message-driven-channel-adapter
    id="my-message-driven-adapter" sqs="sqs" queues="some-queue-of-mine"
    max-number-of-messages="5" visibility-timeout="200" wait-time-out="10"
    send-timeout="2000" channel="inputChannel" />

我观察到消息正确地进入 some-queue-of-mine（通过删除上面的代码并将消息发送到队列）。然后我重新启动我的服务器，启用消息驱动的适配器，我观察到所有消息都从队列中消耗了，但它们去了哪里？ :-/

我的期望是消息将被汇集到一个名为 inputChannel:

的 DirectChannel 中

<int:channel id="inputChannel"/>

我有一个 service-activator 消耗如下：

<int:service-activator ref="myConsumer"
    method='execute' input-channel="inputChannel" output-channel="outputChannel">
    <int:request-handler-advice-chain>
        ...
    </int:request-handler-advice-chain>
</int:service-activator>

当然，我从未见过 myConsumer 被调用。我想我对 MessageProducer 机制如何工作的理解是不够的。有人可以通过提供一个简单的 XML 布线示例来纠正我的想法吗？

Answer 1

根据日志，消息被 handler.AbstractMessageHandler (AbstractMessageHandler.java:115) - ServiceActivator for [org.springframework.integration.handler.MethodInvokingMessageProcessor@493f49cd] 使用。尽管那可能是一个完全不同的故事。

SqsMessageDrivenChannelAdapter 可以与 errorChannel 一起提供以处理下游异常。默认情况下它只被记录。

从该适配器发送的消息如下：

return new GenericMessage<>(message.getBody(), new SqsMessageHeaders(messageHeaders));

其中 message.getBody() 是 String。参见 QueueMessageUtils.createMessage()。

因此，请确保您的 service-activator 接受 String 作为有效载荷，而不是任何其他类型。

如何在 spring-integration-aws 中使用 sqs-message-driven-channel-adapter

How do I use the sqs-message-driven-channel-adapter in spring-integration-aws

spring-integration

spring-cloud-aws