AWS SQS 侦听器无法反序列化消息
AWS SQS Listerner Unable to deserialize the message
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'
or 'false')
at [Source: (String)"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)
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:1913)
at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:772)
at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4340)
at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4189)
at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3205)
at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:230)
...
MessageConversionException: Could not read JSON: Unrecognized token 's3': was expecting (JSON String, Number, Array, Object or token
'null', 'true' or 'false')
at [Source: (String)"s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf";
line: 1, column: 3]; nested exception is
com.fasterxml.jackson.core.JsonParseException: Unrecognized token
's3': was expecting (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";
line: 1, column: 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"
}
您的错误表明问题出在您消息的 Message 键上。
您分享的消息看起来很像 comes from SNS，这是一种常见模式，其中生产者将消息发布到 SNS 主题中，SNS 主题将其消息发布到 SQS 队列中，而消费者则在另一端是 triggered/polls 消息队列。
在您的情况下，正在发生的事情是您的消费者期望 Message 是 JSON 字符串，例如 "{\"some_key\": \"some_value\"}" 因此在读取 Message 的值之后在封装的消息中键入它会尝试将其解析为实际的 dictionary/object.
您应该指示您的代码将此值视为实际字符串并避免转换，或者以 JSON 格式包含您的消息。
通过对消息侦听器进行以下更改来实现此功能
public void receiveInvoice(@NotificationMessage String message) {
//
}
并删除了这一行 messageConverter.setSerializedPayloadClass(String.class);来自 InvoiceSqsConfig.java
的 Bean
@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;
}
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'or 'false') at [Source: (String)"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) at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:1913) at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:772) at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4340) at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4189) at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3205) at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:230) ... MessageConversionException: Could not read JSON: Unrecognized token 's3': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false') at [Source: (String)"s3://abc-invoices/www-abc-at/2020/09/07/504000101-3436547667.pdf"; line: 1, column: 3]; nested exception is com.fasterxml.jackson.core.JsonParseException: Unrecognized token 's3': was expecting (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"; line: 1, column: 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"
}
您的错误表明问题出在您消息的 Message 键上。
您分享的消息看起来很像 comes from SNS，这是一种常见模式，其中生产者将消息发布到 SNS 主题中，SNS 主题将其消息发布到 SQS 队列中，而消费者则在另一端是 triggered/polls 消息队列。
在您的情况下，正在发生的事情是您的消费者期望 Message 是 JSON 字符串，例如 "{\"some_key\": \"some_value\"}" 因此在读取 Message 的值之后在封装的消息中键入它会尝试将其解析为实际的 dictionary/object.
您应该指示您的代码将此值视为实际字符串并避免转换，或者以 JSON 格式包含您的消息。
通过对消息侦听器进行以下更改来实现此功能
public void receiveInvoice(@NotificationMessage String message) {
//
}
并删除了这一行 messageConverter.setSerializedPayloadClass(String.class);来自 InvoiceSqsConfig.java
的 Bean@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;
}