如何在 Spring Boot 中从 AMQP 的 Json 字节数组转换?

2022-01-11 00:00:00 mqtt rabbitmq kotlin java spring-amqp

按照这个，我有这个代码

@Bean
open fun messageConverter(om: ObjectMapper): MessageConverter {
    return Jackson2JsonMessageConverter(om)
}

@Bean
open fun rabbitListenerContainerFactory(cf: ConnectionFactory, mc: MessageConverter ): SimpleRabbitListenerContainerFactory {
    val factory = SimpleRabbitListenerContainerFactory()
    factory.setConnectionFactory(cf)
    factory.setMessageConverter(mc)
    return factory
}

@Bean
open fun builder(): Jackson2ObjectMapperBuilderCustomizer {
    return Jackson2ObjectMapperBuilderCustomizer {
        it.modules(JavaTimeModule(), KotlinModule())
        it.featuresToDisable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)
    }
}

但我遇到了这个异常

org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public phg.entity.monitor.Foley phg.event.monitor.Provisioning.save(phg.entity.monitor.Foley)]
Bean [phg.event.monitor.Provisioning@2212e291]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:193) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:127) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1547) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1473) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1461) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1456) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1405) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:870) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:854) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:78) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1137) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1043) [spring-rabbit-2.1.4.RELEASE.jar:2.1.4.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 [phg.entity.monitor.Foley] for GenericMessage [payload=byte[263], headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=foley.new, amqp_receivedExchange=amq.topic, x-mqtt-publish-qos=1, x-mqtt-dup=false, amqp_deliveryTag=1, amqp_consumerQueue=foley.new, amqp_redelivered=false, id=c7fade5a-d803-166a-fe71-0a4f8abb8c4e, amqp_consumerTag=amq.ctag-9bkLtT_xhvQACuHqvx_izQ, timestamp=1551886588001}]
    at org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver.resolveArgument(PayloadArgumentResolver.java:144) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:117) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-5.1.5.RELEASE.jar:5.1.5.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:49) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:190) ~[spring-rabbit-2.1.4.RELEASE.jar:2.1.4.RELEASE]
    ... 12 common frames omitted

在我注意到它正在使用 GenericMessageConverter 并且没有 Jackson2JsonMessageConverter

stepping through the code before I notice that it's using the GenericMessageConverter and that there isn't a Jackson2JsonMessageConverter

推荐答案

根据您的堆栈跟踪，Listener Container 中与传入的 JSON 没有任何关系.因此，它发生在 pass through 与 byte[] 作为一个 palod.您最终只是尝试将 byte[] 转换为 @RabbitListener 方法上的预期类型.为此，当您没有要在侦听器容器级别转换的 JSON 信息时，您需要使用所需的 MappingJackson2MessageConverter 配置 DefaultMessageHandlerMethodFactory.可以通过RabbitListenerConfigurer来实现:

According your stack trace there is nothing in the Listener Container to do with your incoming JSON. So, it happens as just pass through with byte[] as a pyalod. What you end up is just an attempt to convert that byte[] to the expected type on your @RabbitListener method. For that purpose, when you don't have a JSON information to convert on the listener container level, you need to configure a DefaultMessageHandlerMethodFactory with desired MappingJackson2MessageConverter. It can be achieve with the RabbitListenerConfigurer:

    @Bean
    public DefaultMessageHandlerMethodFactory myHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(new MappingJackson2MessageConverter());
        return factory;
    }


    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(myHandlerMethodFactory());
    }

MappingJackson2MessageConverter 可以配置为您的自定义 ObjectMapper 注入的 bean.

That MappingJackson2MessageConverter can be configured as a bean for your custom ObjectMapper injection.

相关文章