时间戳

首页 > 解决方案 > AWS SQS Listerner 无法反序列化消息

amazon-web-services - AWS SQS Listerner 无法反序列化消息

问题描述

Jackson 无法反序列化它。请找到堆栈跟踪和 SQS 配置和侦听器

org.springframework.cloud.aws.messaging.listener.QueueMessageHandler.processHandlerMethodException(QueueMessageHandler.java:248)
      t stack_trace   JsonParseException: Unrecognized token 's3': was expecting (JSON String, Number, Array, Object or token 'null', 'true'

或“假”）在[来源：（字符串）“s3：//abc-invoices45-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf”；line: 1, column: 3] at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1840) at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:722) at com .fasterxml.jackson.core.json.ReaderBasedJsonParser._reportInvalidToken(ReaderBasedJsonParser.java:2867) 在 com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:1913) 在 com.fasterxml.jackson.core.json .ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:772) 在 com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4340) 在 com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4189) 在com.fasterxml.jackson.databind.ObjectMapper。readValue(ObjectMapper.java:3205) at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:230) ... MessageConversionException：无法读取 JSON：无法识别的令牌“s3”：期待（JSON 字符串，数字，数组，对象或令牌'null'，'true'或'false'）在[来源：（字符串）“s3：//abc-invoices/www-abc-at/2020/09/07/504000101-3436547667。 pdf"; 行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:230) ... MessageConversionException：无法读取 JSON：无法识别的令牌 's3'：期待（JSON 字符串、数字、数组、对象或令牌 'null'、' true' 或 'false'）在 [来源：（字符串）“s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf”；行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:230) ... MessageConversionException：无法读取 JSON：无法识别的令牌 's3'：期待（JSON 字符串、数字、数组、对象或令牌 'null'、' true' 或 'false'）在 [来源：（字符串）“s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf”；行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] convertFromInternal(MappingJackson2MessageConverter.java:230) ... MessageConversionException：无法读取 JSON：无法识别的令牌“s3”：期待（JSON 字符串、数字、数组、对象或令牌“null”、“true”或“false”） [来源：（字符串）“s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf”；行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] convertFromInternal(MappingJackson2MessageConverter.java:230) ... MessageConversionException：无法读取 JSON：无法识别的令牌“s3”：期待（JSON 字符串、数字、数组、对象或令牌“null”、“true”或“false”） [来源：（字符串）“s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf”；行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] [来源：（字符串）“s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf "; 行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] [来源：（字符串）“s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf "; 行：1，列：3]；嵌套异常是 com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expected (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String )"s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf"; 行：1，列：3] [来源：（字符串）“s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf”中的对象或令牌“空”、“真”或“假”）；行：1，列：3] [来源：（字符串）“s3://abc-invoices/www-k24-at/2020/09/07/504000101-3436547667.pdf”中的对象或令牌“空”、“真”或“假”）；行：1，列：3]

这是我的 SQS 配置



import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.sqs.AmazonSQSAsync;
import com.amazonaws.services.sqs.AmazonSQSAsyncClientBuilder;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.abc.properties.LocalPropertyService;
import java.util.Collections;
import org.springframework.cloud.aws.messaging.config.QueueMessageHandlerFactory;
import org.springframework.cloud.aws.messaging.config.SimpleMessageListenerContainerFactory;
import org.springframework.cloud.aws.messaging.config.annotation.SqsConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;

@Configuration
@Import(SqsConfiguration.class)
public class InvoiceSqsConfig {

  @Bean(name = "amazonSQSAsyncClient")
  public AmazonSQSAsync getAmazonSQSAsyncClient(LocalPropertyService propertyService) {
    System.setProperty("invoice.queue.name",
                       propertyService.getString("sqs.invoice.queueName"));
    return AmazonSQSAsyncClientBuilder.standard()
                                      .withRegion(
                                          propertyService.getString("sqs.aws.region"))
                                      .withCredentials(
                                          new AWSStaticCredentialsProvider(new BasicAWSCredentials(
                                              propertyService.getString("sqs.invoice.accessKey"),
                                              propertyService.getString("sqs.invoice.secretKey"))))
                                      .build();
  }

  @Bean
  public QueueMessageHandlerFactory queueMessageHandlerFactory() {
    QueueMessageHandlerFactory factory = new QueueMessageHandlerFactory();
    MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
    messageConverter.setStrictContentTypeMatch(false);
    messageConverter.getObjectMapper().registerModule(new JavaTimeModule());
    factory.setMessageConverters(Collections.singletonList(messageConverter));
    return factory;
  }

  @Bean
  public SimpleMessageListenerContainerFactory simpleMessageListenerContainerFactory(AmazonSQSAsync amazonSqs) {
    SimpleMessageListenerContainerFactory factory = new SimpleMessageListenerContainerFactory();
    factory.setAmazonSqs(amazonSqs);
    factory.setMaxNumberOfMessages(10);
    factory.setWaitTimeOut(20);
    return factory;
  }
}

这是消息监听器

import static org.springframework.cloud.aws.messaging.listener.SqsMessageDeletionPolicy.ON_SUCCESS;

import com.abc.order.invoice.exception.InvalidInvoiceMessageException;
import com.abc.order.invoice.exception.OrderNotFoundException;
import com.abc.order.order.OrderService;
import com.abc.order.order.model.Order;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.cloud.aws.messaging.config.annotation.NotificationMessage;
import org.springframework.cloud.aws.messaging.listener.annotation.SqsListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
@RequiredArgsConstructor
public class InvoiceUpdateListener {

  @NonNull
  private final OrderService orderService;

  @SqsListener(value = "${invoice.queue.name}", deletionPolicy = ON_SUCCESS)
  @SneakyThrows
  public void receiveInvoice(@NotificationMessage EnvelopedMessage envelopedMessage) {
    log.debug("Received message from the invoice queue : {} ", envelopedMessage.getMessage());
    String message = envelopedMessage.getMessage();
    if (StringUtils.isBlank(message)) {
      throw new InvalidInvoiceMessageException("Received invalid message from the invoice queue");
    }
    String orderNumber = extractOrderNumber(message);
    Order order = orderService.getOrderByOrderNumber(orderNumber);
    if (order == null) {
      throw new OrderNotFoundException(
          "Could not find the order with order number : " + orderNumber);
    }
    order.setInvoiceUrl(message);
    log.debug("Saving invoice url for the orderNumber : {} ", orderNumber);
    orderService.save(order);
  }

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@JsonInclude(Include.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown = true)
public class EnvelopedMessage {

  @JsonProperty("Type")
  private String type;
  @JsonProperty("MessageId")
  private String messageId;
  @JsonProperty("TopicArn")
  private String topicArn;
  @JsonProperty("Subject")
  private String subject;
  @JsonProperty("Message")
  private String message;
  @JsonProperty("Timestamp")
  private ZonedDateTime createdAt;
  @JsonProperty("SignatureVersion")
  private String signatureVersion;
  @JsonProperty("Signature")
  private String signature;
  @JsonProperty("SigningCertURL")
  private String certUrl;
  @JsonProperty("UnsubscribeURL")
  private String unsubscribeUrl;
}

这是我从生产者那里收到的消息

{
  "Type" : "Notification",
  "MessageId" : "d77fa67e-6aa6-5b87-a707-f1ae5ba5922f",
  "TopicArn" : "arn:aws:sns:eu-central-1:726569450381:invoices-from-core",
  "Subject" : "File uploaded: invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf",
  "Message" : "s3://k24-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf",
  "Timestamp" : "2020-09-07T12:59:47.192Z",
  "SignatureVersion" : "1",
  "Signature" : "dummysignature",
  "SigningCertURL" : "dummy url",
  "UnsubscribeURL" : "dummy url"
}

标签： amazon-web-servicesjacksonamazon-sqsmessage-listener

解决方案

您的错误表明问题在于Message您的消息的关键。

您共享的消息看起来实际上来自 SNS，这是一种常见模式，其中生产者将消息发布到 SNS 主题中，SNS 主题将其消息发布到 SQS 队列中，另一端的消费者被触发/轮询消息队列。

在您的情况下，正在发生的事情是您的消费者期望它Message是一个 JSON 字符串"{\"some_key\": \"some_value\"}"，在读取Message封装消息中的键值后，它会尝试将其解析为实际的字典/对象。

您应该指示您的代码将此值视为实际字符串并避免转换，或者将您的消息包含在 JSON 格式中。

推荐阅读