将 SqsListener 与 SNS 和 SQS 一起使用

Question

我正在使用 spring-cloud-aws 的 SqsListener 从 AWS 的简单队列服务 (SQS) 接收 JSON Format 中的 AWS SNS HTTP 通知。

这是监听器的代码：

@SqsListener(value = "my-queue", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void handle(final MyObject obj) throws Exception {
// ...
}

上面链接的文档只是关于向队列发送和读取普通序列化对象，我认为接收 SNS 消息应该是开箱即用的。但我最终收到转换错误：

10:45:51.480 [simpleMessageListenerContainer-2] 错误 o.s.c.a.m.l.SimpleMessageListenerContainer - 遇到异常 在处理消息时。 org.springframework.messaging.MessagingException：一个异常 调用处理程序方法时发生；嵌套异常是 org.springframework.messaging.converter.MessageConversionException：否 转换器发现转换为 com.myproject.model.MyObject 类， 消息=通用消息

我还尝试创建一个看起来像上面链接的预期 SNS Json 格式的包装器对象，但我不断收到相同的异常。唯一有效的类型是签名中的字符串。 SNS不应该自动转换吗？

Answer 1

是的，应该。它确实如此。

为了在反序列化时调用正确的HandlerMethodArgumentResolver（在本例中为NotificationMessageArgumentResolver），进而调用正确的转换器NotificationRequestConverter，您只需将注释org.springframework.cloud.aws.messaging.config.annotation.NotificationMessage 添加到您的方法签名中。例如

@SqsListener(value = "my-queue", deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void handle(final @NotificationMessage MyObject obj) throws Exception {
// ...
}

这样，您的 SNS 的 Message 部分将被提取并转换为 MyObject。

Answer 2

这在没有@NotificationMessage 的情况下也有效。这样您就不需要发送使用此注释所需的“类型”和“消息”部分。

首先创建一个具有所需属性的类。

public class SqsMessage {

   private String myTask;

   public SqsMessage() {
   }

   public SqsMessage(@JsonProperty("MyTask") String myTask ) {
       this.myTask = myTask ;
   }

   //Getter + Setter 
}

接下来设置监听器

@SqsListener(value = {"MyQueue"}, deletionPolicy = SqsMessageDeletionPolicy.ON_SUCCESS)
public void receiveMessage(SqsMessage payload, @Headers Map<String, Object> header) {
   logger.info("Got message with task: " + payload.getTask() 
    + " with custom attribute " + header.get("CustomAttribute").toString());
}

现在你可以发送 JSON 格式

{"MyTask":"My task"}

POJO 构造函数中的@JsonProperty("MyTask") 注释可以是可选的，具体取决于您的 spring 版本以及您的属性是否与 Json 字符串中的名称相同。例如，如果您的属性名为 task 而您的 Json 字符串为 {"task":"My task"}，则没有必要。