Rabbit消费对象天坑（序列化/反序列化）_error handler converted exception to fatal

作者：凡人多烦事01 | 2024-05-26 08:22:52

踩

error handler converted exception to fatal

一、问题描述


2022-05-03 14:01:40.630  WARN 16876 --- [ntContainer#0-2] s.a.r.l.ConditionalRejectingErrorHandler : Execution of Rabbit message listener failed.
 
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Failed to convert message
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:155) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1665) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1584) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1572) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1563) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1507) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:967) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1291) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1197) [spring-rabbit-2.4.2.jar:2.4.2]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_301]
Caused by: org.springframework.amqp.support.converter.MessageConversionException: failed to convert serialized Message content
    at org.springframework.amqp.support.converter.SimpleMessageConverter.fromMessage(SimpleMessageConverter.java:117) ~[spring-amqp-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener.extractMessage(AbstractAdaptableMessageListener.java:342) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter$MessagingMessageConverterAdapter.extractPayload(MessagingMessageListenerAdapter.java:365) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.support.converter.MessagingMessageConverter.fromMessage(MessagingMessageConverter.java:132) ~[spring-amqp-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.toMessagingMessage(MessagingMessageListenerAdapter.java:242) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:145) ~[spring-rabbit-2.4.2.jar:2.4.2]
    ... 11 common frames omitted
Caused by: java.lang.IllegalStateException: Could not deserialize object type
    at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:97) ~[spring-amqp-2.4.2.jar:2.4.2]
    at org.springframework.amqp.support.converter.SimpleMessageConverter.fromMessage(SimpleMessageConverter.java:113) ~[spring-amqp-2.4.2.jar:2.4.2]
    ... 16 common frames omitted
Caused by: java.lang.ClassNotFoundException: com.example.cloud.producer.rabbitmq.entity.RabbitOrder
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382) ~[na:1.8.0_301]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418) ~[na:1.8.0_301]
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355) ~[na:1.8.0_301]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ~[na:1.8.0_301]
    at java.lang.Class.forName0(Native Method) ~[na:1.8.0_301]
    at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_301]
    at org.springframework.util.ClassUtils.forName(ClassUtils.java:284) ~[spring-core-5.3.16.jar:5.3.16]
    at org.springframework.core.ConfigurableObjectInputStream.resolveClass(ConfigurableObjectInputStream.java:76) ~[spring-core-5.3.16.jar:5.3.16]
    at org.springframework.amqp.support.converter.SimpleMessageConverter$1.resolveClass(SimpleMessageConverter.java:183) ~[spring-amqp-2.4.2.jar:2.4.2]
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1984) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1848) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2158) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1665) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:501) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:459) ~[na:1.8.0_301]
    at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:91) ~[spring-amqp-2.4.2.jar:2.4.2]
    ... 17 common frames omitted
 
2022-05-03 14:01:40.630  WARN 16876 --- [ntContainer#0-2] ingErrorHandler$DefaultExceptionStrategy : Fatal message conversion error; message rejected; it will be dropped or routed to a dead letter exchange, if so configured: (Body:'[serialized object]' MessageProperties [headers={spring_listener_return_correlation=b2329a07-f83e-4999-b7b4-5d1c7b65c767, spring_returned_message_correlation=1651557699966$3b0f387d-b17f-42df-b060-06ffa0a71910}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=order-exchange, receivedRoutingKey=order.user, deliveryTag=1, consumerTag=amq.ctag-sxIMscspKJ98oEOaFyIyPQ, consumerQueue=order-queue])
2022-05-03 14:01:40.630 ERROR 16876 --- [ntContainer#0-2] o.s.a.r.l.SimpleMessageListenerContainer : Execution of Rabbit message listener failed, and the error handler threw an exception
 
org.springframework.amqp.AmqpRejectAndDontRequeueException: Error Handler converted exception to fatal
    at org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler.handleError(ConditionalRejectingErrorHandler.java:146) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:1469) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:1753) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1528) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:967) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:914) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$1600(SimpleMessageListenerContainer.java:83) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1291) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1197) [spring-rabbit-2.4.2.jar:2.4.2]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_301]
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Failed to convert message
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:155) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1665) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1584) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1572) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1563) [spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1507) [spring-rabbit-2.4.2.jar:2.4.2]
    ... 6 common frames omitted
Caused by: org.springframework.amqp.support.converter.MessageConversionException: failed to convert serialized Message content
    at org.springframework.amqp.support.converter.SimpleMessageConverter.fromMessage(SimpleMessageConverter.java:117) ~[spring-amqp-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener.extractMessage(AbstractAdaptableMessageListener.java:342) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter$MessagingMessageConverterAdapter.extractPayload(MessagingMessageListenerAdapter.java:365) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.support.converter.MessagingMessageConverter.fromMessage(MessagingMessageConverter.java:132) ~[spring-amqp-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.toMessagingMessage(MessagingMessageListenerAdapter.java:242) ~[spring-rabbit-2.4.2.jar:2.4.2]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:145) ~[spring-rabbit-2.4.2.jar:2.4.2]
    ... 11 common frames omitted
Caused by: java.lang.IllegalStateException: Could not deserialize object type
    at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:97) ~[spring-amqp-2.4.2.jar:2.4.2]
    at org.springframework.amqp.support.converter.SimpleMessageConverter.fromMessage(SimpleMessageConverter.java:113) ~[spring-amqp-2.4.2.jar:2.4.2]
    ... 16 common frames omitted
Caused by: java.lang.ClassNotFoundException: com.example.cloud.producer.rabbitmq.entity.RabbitOrder
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382) ~[na:1.8.0_301]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418) ~[na:1.8.0_301]
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355) ~[na:1.8.0_301]
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ~[na:1.8.0_301]
    at java.lang.Class.forName0(Native Method) ~[na:1.8.0_301]
    at java.lang.Class.forName(Class.java:348) ~[na:1.8.0_301]
    at org.springframework.util.ClassUtils.forName(ClassUtils.java:284) ~[spring-core-5.3.16.jar:5.3.16]
    at org.springframework.core.ConfigurableObjectInputStream.resolveClass(ConfigurableObjectInputStream.java:76) ~[spring-core-5.3.16.jar:5.3.16]
    at org.springframework.amqp.support.converter.SimpleMessageConverter$1.resolveClass(SimpleMessageConverter.java:183) ~[spring-amqp-2.4.2.jar:2.4.2]
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1984) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1848) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2158) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1665) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:501) ~[na:1.8.0_301]
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:459) ~[na:1.8.0_301]
    at org.springframework.amqp.utils.SerializationUtils.deserialize(SerializationUtils.java:91) ~[spring-amqp-2.4.2.jar:2.4.2]
    ... 17 common frames omitted

二、解决方案

方案一：共同使用一个对象

描述：我目前生产者和消费者都有一个对象，内容是一模一样的包括序列化，可就是消费者接收时报序列化错误

解决方案：把这个对象放到公共模块，然后生产者和消费者服务引入这个公共模块，生产者和消费者发送和消费时共用一个对象。这样完全保证了两个项目中JavaBean是一致的，所以能解决反序列失败的问题

缺点：局限性太小，这种模式生产者和消费者只能在同一个微服务下才能使用

方案二：消息JSON序列化（推荐）

生产者确认机制配置


package com.vueadmin.security.config;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import java.util.HashMap;
 
@Configuration
public class RabbitConfig {
 
    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;
 
    private final String EXCHANGE_NAME="nome_exchange";
 
//    private final String DeadEXCHANGE_NAME="dead_nome_exchange";
 
    private final String QUEUE_NAME="nome_queue";
//    private final String DeadQUEUE_NAME="dead_nome_queue";
 
 
    @Bean
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
        rabbitTemplate.setConfirmCallback((data,ask,cause)->{
            if (ask){
                System.out.println("发送交换机成功"+data.getId());
            }else
                System.out.println("发送交换机失败"+data.getId());
        });
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnsCallback((data)->{
            System.out.println("发送队列失败"+new String(data.getMessage().getBody()));
        });
 
        return rabbitTemplate;
    }
 
//    @Bean
//    public RabbitTemplate rabbitTemplate(){
//        RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
//        rabbitTemplate.setConfirmCallback((data,ask,cause)->{
//            if (ask){
//                System.out.println("发送交换机成功"+data.getId());
//                修改库中消息状态
//                updateStatusById(data.getId());
//            }else
//                System.out.println("发送交换机失败"+data.getId());
//        });
//        rabbitTemplate.setMandatory(true);
//        rabbitTemplate.setReturnsCallback((data)->{
//            System.out.println("发送队列失败"+new String(data.getMessage().getBody()));
//        });
//
//        return rabbitTemplate;
//    }
 
    @Bean("nome_queue")
    public Queue queue(){
//        HashMap<String, Object> map = new HashMap<>();
//        map.put("x-dead-letter-exchange",DeadEXCHANGE_NAME);
//        map.put("x-max-length",5);
//        map.put("x-message-ttl",5000);
//        map.put("x-message-ttl",50000);
 
        return new Queue(QUEUE_NAME,true,false,false);
    }
 
//    @Bean("dead_nome_queue")
//    public Queue deadQueue(){
//        return new Queue(DeadQUEUE_NAME,true,false,false);
//    }
 
 
    @Bean("nome_exchange")
    public DirectExchange directExchange(){
        return new DirectExchange(EXCHANGE_NAME);
    }
 
//    @Bean("dead_nome_exchange")
//    public DirectExchange deadDirectExchange(){
//        return new DirectExchange(DeadEXCHANGE_NAME);
//    }
 
    @Bean
    public Binding binding(){
        return BindingBuilder.bind(queue()).to(directExchange()).with("nome");
    }
 
//    @Bean
//    public Binding deadBinding(){
//        return BindingBuilder.bind(deadQueue()).to(deadDirectExchange()).with("nome");
//    }
}

yml配置


spring:
    rabbitmq:
        host: 192.168.72.128
        virtual-host: /emp
        username: user
        password: 123456
        port: 5672
        publisher-returns: true
        listener:
            simple:
                acknowledge-mode: manual

生产者序列化


package com.vueadmin.security.config;
 
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class RabbitmqConfig implements InitializingBean {
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
    @Override
    public void afterPropertiesSet() throws Exception {
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
    }
}


 @Test
    void TextDirect(){
 
        CorrelationData data = new CorrelationData();
        String s = UUID.randomUUID().toString();
        data.setId(s);
//        MessagePostProcessor message = new MessagePostProcessor() {
//            @Override
//            public Message postProcessMessage(Message message) throws AmqpException {
//                message.getMessageProperties().setExpiration("5000");
//                return message;
//            }
//        };
        Emp emp = new Emp();
        emp.setId(1);
        emp.setName("张三");
//        String str = JSONUtil.toJsonStr(employee);
 
        rabbitTemplate.convertAndSend("nome_exchange","nome",emp,data);
    }

消费者反序列化


@Bean
    public MessageConverter jsonMessageConverter(ObjectMapper objectMapper) {
        return new Jackson2JsonMessageConverter(objectMapper);
    }

消费者接收消息


package com.test.csfl.pojo;
 
 
import com.rabbitmq.client.Channel;
import com.test.csfl.entity.Emp;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
 
import java.io.IOException;
import java.util.Map;
 
@Component
public class Consume {
 
 
    @RabbitListener(queues = "nome_queue")
    public void Handler(@Payload Emp emp, @Headers Map<String,Object> headers,Channel channel) throws IOException {
 
 
        System.out.println(emp.getName());
        long o = (long) headers.get(AmqpHeaders.DELIVERY_TAG);
        String s = (String) headers.get("spring_returned_message_correlation");
        System.out.println(s);
        channel.basicAck(o,false);
    }
}

根据此文章信息进行书写

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/凡人多烦事01/article/detail/625950