spring-integration - 我如何在 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 作为 paylaod 而不是任何其他类型。