java - 如何在java中从AWS-SQS读取反序列化对象？

Question

我想从 springboot 应用程序读取 SNS->SQS 消息 ({"author":"Peter Smith","error":"error of test"})，但我不反序列化我的对象。我有这个错误:

ERROR 1484 --- [enerContainer-2] o.s.c.a.m.listener.QueueMessageHandler   : Unhandled exception from message handler method

org.springframework.messaging.converter.MessageConversionException: Could not read JSON: Cannot construct instance of `com.saltandpepper.event.domain.event.dto.PushEvent` (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value ('{"author":"Peter Smith","error":"error of test"}')
 at [Source: (String)"{
  "Type" : "Notification",
  "MessageId" : "56639e45-586e-5281-9204-cbeed1890b81",
  "TopicArn" : "arn:aws:sns:eu-west-1:xxxxxxxxxxxx:eventNow",
  "Message" : "{\"author\":\"Peter Smith\",\"error\":\"error of test\"}",
  "Timestamp" : "2019-07-19T09:04:39.109Z",
  "SignatureVersion" : "1",
  "Signature" : "hnSn2gWYqmnpmgGpqGy8/4V5mTirHEVB5Bmar3AnGayevXMqV+rk8CRbEcyOmoao7krUctnvMwQKubVDQQFKszgQZ7qxdEp7I0cJobiipdwM6z7D4O+pvJJ3pyjxGYQNcNLF1rvFJoTnSW/uUAdvLYv4vuKeNeF11P58dscKF+HCmAnLMEe6EnZNNzeb"[truncated 724 chars]; line: 5, column: 15] (through reference chain: com.orange.newcp.model.SqsMessage["Message"]); nested exception is com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance of `com.saltandpepper.event.domain.event.dto.PushEvent` (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value ('{"author":"Peter Smith","error":"error of test"}')
 at [Source: (String)"{
  "Type" : "Notification",
  "MessageId" : "56639e45-586e-5281-9204-cbeed1890b81",
  "TopicArn" : "arn:aws:sns:eu-west-1:xxxxxxxxxxxx:eventNow",
  "Message" : "{\"author\":\"Peter Smith\",\"error\":\"error of test\"}",
  "Timestamp" : "2019-07-19T09:04:39.109Z",
  "SignatureVersion" : "1",
  "Signature" : "hnSn2gWYqmnpmgGpqGy8/4V5mTirHEVB5Bmar3AnGayevXMqV+rk8CRbEcyOmoao7krUctnvMwQKubVDQQFKszgQZ7qxdEp7I0cJobiipdwM6z7D4O+pvJJ3pyjxGYQNcNLF1rvFJoTnSW/uUAdvLYv4vuKeNeF11P58dscKF+HCmAnLMEe6EnZNNzeb"[truncated 724 chars]; line: 5, column: 15] (through reference chain: com.orange.newcp.model.SqsMessage["Message"])
    at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:234) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.converter.AbstractMessageConverter.fromMessage(AbstractMessageConverter.java:181) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:137) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:117) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:148) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler.handleMatch(AbstractMethodMessageHandler.java:550) [spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler.handleMessageInternal(AbstractMethodMessageHandler.java:505) [spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.messaging.handler.invocation.AbstractMethodMessageHandler.handleMessage(AbstractMethodMessageHandler.java:439) [spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer.executeMessage(SimpleMessageListenerContainer.java:227) [spring-cloud-aws-messaging-2.1.1.RELEASE.jar:2.1.1.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer$MessageExecutor.run(SimpleMessageListenerContainer.java:417) [spring-cloud-aws-messaging-2.1.1.RELEASE.jar:2.1.1.RELEASE]
    at org.springframework.cloud.aws.messaging.listener.SimpleMessageListenerContainer$SignalExecutingRunnable.run(SimpleMessageListenerContainer.java:309) [spring-cloud-aws-messaging-2.1.1.RELEASE.jar:2.1.1.RELEASE]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_201]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_201]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_201]
Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot construct instance of `com.saltandpepper.event.domain.event.dto.PushEvent` (although at least one Creator exists): no String-argument constructor/factory method to deserialize from String value ('{"author":"Peter Smith","error":"error of test"}')
 at [Source: (String)"{
  "Type" : "Notification",
  "MessageId" : "56639e45-586e-5281-9204-cbeed1890b81",
  "TopicArn" : "arn:aws:sns:eu-west-1:xxxxxxxxxxxx:eventNow",
  "Message" : "{\"author\":\"Peter Smith\",\"error\":\"error of test\"}",
  "Timestamp" : "2019-07-19T09:04:39.109Z",
  "SignatureVersion" : "1",
  "Signature" : "hnSn2gWYqmnpmgGpqGy8/4V5mTirHEVB5Bmar3AnGayevXMqV+rk8CRbEcyOmoao7krUctnvMwQKubVDQQFKszgQZ7qxdEp7I0cJobiipdwM6z7D4O+pvJJ3pyjxGYQNcNLF1rvFJoTnSW/uUAdvLYv4vuKeNeF11P58dscKF+HCmAnLMEe6EnZNNzeb"[truncated 724 chars]; line: 5, column: 15] (through reference chain: com.orange.newcp.model.SqsMessage["Message"])
    at com.fasterxml.jackson.databind.exc.MismatchedInputException.from(MismatchedInputException.java:63) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.DeserializationContext.reportInputMismatch(DeserializationContext.java:1343) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.DeserializationContext.handleMissingInstantiator(DeserializationContext.java:1032) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.ValueInstantiator._createFromStringFallbacks(ValueInstantiator.java:371) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.std.StdValueInstantiator.createFromString(StdValueInstantiator.java:323) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromString(BeanDeserializerBase.java:1373) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeOther(BeanDeserializer.java:171) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:161) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:129) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:369) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:159) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4013) ~[jackson-databind-2.9.9.jar:2.9.9]
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3042) ~[jackson-databind-2.9.9.jar:2.9.9]
    at org.springframework.messaging.converter.MappingJackson2MessageConverter.convertFromInternal(MappingJackson2MessageConverter.java:229) ~[spring-messaging-5.1.8.RELEASE.jar:5.1.8.RELEASE]
    ... 14 common frames omitted

我有这个代码:

我的 Springboot 监听器:

@SqsListener(QUEUE_NAME)
public void receiveMessage(SqsMessage message, @Header("SenderId") String senderId) {
    logger.info("Service A received message: {}", message);
    logger.info("Service A received messageId: {}", message.getMessageId());
    logger.info("Service A received message: {}", message.getMessage());
    logger.info("Service A received author: {}", message.getMessage().getAuthor());
    logger.info("Service A senderId: {}", senderId);
}

我的第一个 POJO:

public class SqsMessage {

    @JsonProperty("MessageId")
    String messageId;

    @JsonProperty("Message")
    PushEvent<?> message;

    public SqsMessage() {
    }

    public SqsMessage(String messageId, PushEvent<?> message) {
       this.messageId = messageId ;
       this.message = message ;
    }

    public String getMessageId() {
        return messageId;
    }

    public void setMessageId(String messageId) {
        this.messageId = messageId;
    }

    public PushEvent<?> getMessage() {
        return message;
    }

    public void setMessage(PushEvent<?> message) {
        this.message = message;
    }

}

我的第二个 POJO:

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class PushEvent<T> {
    String author;
    // T current;
    String error;
}

我的 Springboot 配置:

@Configuration
public class SpringCloudAwsConfig {

    @Bean
    public QueueMessageHandlerFactory queueMessageHandlerFactory() {
        QueueMessageHandlerFactory factory = new QueueMessageHandlerFactory();
        MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();

        //set strict content type match to false
        messageConverter.setStrictContentTypeMatch(false);
        factory.setArgumentResolvers(Collections.<HandlerMethodArgumentResolver>singletonList(new PayloadArgumentResolver(messageConverter)));
        List<MessageConverter> mc = new ArrayList<>();
        mc.add(new MappingJackson2MessageConverter());
        factory.setMessageConverters(mc);
        return factory;
    }

    @Bean
    public QueueMessagingTemplate queueMessagingTemplate(AmazonSQSAsync amazonSQSAsync) {
        return new QueueMessagingTemplate(amazonSQSAsync);
    }

    @Bean
    public NotificationMessagingTemplate notificationMessagingTemplate(AmazonSNS amazonSNS) {
        return new NotificationMessagingTemplate(amazonSNS);
    }

}

Answer 1

我将任何自定义反序列化器添加到映射器

SimpleModule module = new SimpleModule();
module.addDeserializer(Rate.class, new RateDeserializer());
SimpleModule module2 = new SimpleModule();
module2.addDeserializer(Maturity.class, new MaturityDeserializer());
ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule()).registerModule(module).registerModule(module2);
mapper.setDateFormat(new StdDateFormat());

我使用这个映射器:

PushEventMessage<PushEvent<Offer>> pushEventMessage = mapper.readValue(message.getMessage(), PushEventMessage.class);

自定义解串器:

public class RateDeserializer extends StdDeserializer<Rate> {

    static final String VALUE_NODE_NAME = "value";
    static final String RATESUBTYPE_NODE_NAME = "rateSubType";

    public RateDeserializer() {
        super(Rate.class);
    }

    @Override
    public Rate deserialize(final JsonParser jsonParser, final DeserializationContext deserializationContext) throws IOException {

        ...

    }
}