java - SpringBoot RabbitMq解析盒子的json消息auto

Question

我的 springboot 应用程序连接到 RabbitMQ 队列，在其中以指定的 Json 格式推送消息，例如: {"messageId":"de5fe5a3-1831-4b87-891d-e7a4c29295b4","message": {"listingId":"2"},"errors":[]} 我确实创建了相应的Java DTO 对象，但我不知道如何将rabbbitMq 消息直接解析为java 对象。我已经找到了一种使用 Jackson ObjectMapper 来做到这一点的方法，但是我会就像 spring 自己做而不是我

我的 DTO

    @Data
    @JsonIgnoreProperties(ignoreUnknown = true)
    final class ListingMessage {
        private final String messageId;
        private final Message message;
        private final List<String> errors;

        @Data
        static final class Message {
            private final String listingId;
        }
    }

我想实现这样的目标:

    @Component
    @Slf4j
    final class ListingMessageListener {

        private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

        @RabbitListener(queues = "${rabbitmq.queues}")
        public void receiveMessage(final ListingMessage message) {
                doSomeStuff(message); 
         }
       }

而不是

@Component
@Slf4j
final class ListingMessageListener {

    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    @RabbitListener(queues = "${rabbitmq.queues}")
    public void receiveMessage(final Message message) {

        final String json = new String(message.getBody());
        try {
            final ListingMessage listingMessage = OBJECT_MAPPER.readValue(json, ListingMessage.class);
            doSomeStuff(listingMessage);
        }catch(final Exception e){e.printStackTrace();}
     }
   }

编辑。当我添加 Jackson2JsonMessageConverter

    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }

我遇到了异常

org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.listingsdatapipeline.rabbitmq.ListingMessageListener.receiveMessage(com.listingsdatapipeline.rabbitmq.ListingMessage)]
Bean [com.listingsdatapipeline.rabbitmq.ListingMessageListener@626df4e5]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:191) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:126) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1414) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1337) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1324) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1303) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:817) [spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:801) [spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:77) [spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1042) [spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.listingsdatapipeline.rabbitmq.ListingMessage] for GenericMessage [payload=byte[167], headers={amqp_receivedDeliveryMode=NON_PERSISTENT, amqp_receivedRoutingKey=changes-consumer-de, amqp_deliveryTag=1, amqp_consumerQueue=changes-consumer-de, content-type=application/json, amqp_redelivered=false, id=8fb1f2ad-96dd-3d25-c77d-82f741cfa788, amqp_consumerTag=amq.ctag-DfMMARR_KXMXyNJMoEXyAQ, timestamp=1539177836655}]
    at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:144) ~[spring-messaging-5.0.9.RELEASE.jar:5.0.9.RELEASE]
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:116) ~[spring-messaging-5.0.9.RELEASE.jar:5.0.9.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:137) ~[spring-messaging-5.0.9.RELEASE.jar:5.0.9.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:109) ~[spring-messaging-5.0.9.RELEASE.jar:5.0.9.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:51) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:188) ~[spring-rabbit-2.0.6.RELEASE.jar:2.0.6.RELEASE]
    ... 10 common frames omitted

Answer 1

只需将 Jackson2JsonMessageConverter @Bean 添加到您的应用程序配置中，Boot 将自动配置监听器容器来使用它；然后，@RabbitListener 适配器将尝试转换为 ListingMessage，因为这是方法参数类型。