RabbitMQ Listener 在 Java 17 和 Spring boot 3 升级后抛出致命异常

Question

我将项目从 Java 11/Spring Boot 2.7.9 升级到 Java 17/Spring Boot 3.1.3/spring-rabbit 3.0.8。发送方仍然可以将对象发送到 RabbitMQ，但侦听器会抛出异常，“未从实际负载类型中找到转换器”。

---

1.RabbitMQ 发送方：

private void enQueue(ScannerPub pub) throws JsonProcessingException {
    try {
        this.rabbitTemplate.convertAndSend(
                queueConfig.getTopicExchangeName(), queueConfig.getScannerRoutingkey(), pub);
    } catch (Exception ex) {
        ....
    }
}

---

2.型号：

package ca.bc.gov.open.crdp.process.models;

import java.io.Serializable;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ScannerPub implements Serializable {
    private String filePath;
    private String dateTime;
}

---

3.RabbitMQ监听器：

@RabbitListener(queues = "${crdp.scanner-queue}")
public void receiveScannerPubMessage(@Payload Message<ScannerPub> message) {
    transformerService.processFileService(message.getPayload());
} 

---

4.Exception
---------------------------------------------------------------------------------------------
summary - 1), 2) and 3) are the errors I select from Detail which is too long for reading:

1) org.springframework.messaging.converter.MessageConversionException: No converter found from actual payload type 'ca.bc.gov.open.crdp.process.models.ScannerPub' to expected payload type 'ca.bc.gov.open.crdp.process.models.ScannerPub'
    
2)ERROR org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer: Execution of Rabbit message listener failed, and the error handler threw an exception org.springframework.amqp.AmqpRejectAndDontRequeueException: Error Handler converted exception to fatal
    
3) 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={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=true, receivedExchange=crdp-exchange, receivedRoutingKey=SCANNER, deliveryTag=1, consumerTag=amq.ctag-p37Eiyz_B5hZDTrhkBlk6Q, consumerQueue=scanner-queue])

Detail:------------------------------------------------------------------------------------

2023-10-27 21:51:20,985 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-2] ERROR org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer: Execution of Rabbit message listener failed, and the error 
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:

2023-10-27 21:51:21,001 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-2] WARN  org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler: Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void ca.bc.gov.open.crdp.process.transformer.services.ConsumerService.receiveScannerPubMessage(org.springframework.messaging.Message<ca.bc.gov.open.crdp.process.models.ScannerPub>)]
Bean [ca.bc.gov.open.crdp.process.transformer.services.ConsumerService@74e89b46]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:281)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:224)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:149)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1663)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1582)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1570)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1561)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListenerAndHandleException(AbstractMessageListenerContainer.java:1506)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.lambda$executeListener$8(AbstractMessageListenerContainer.java:1484)
    at io.micrometer.observation.Observation.lambda$observe$0(Observation.java:493)
    at io.micrometer.observation.Observation.observeWithContext(Observation.java:603)
    at io.micrometer.observation.Observation.observe(Observation.java:492)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1484)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:994)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:941)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1323)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1225)
    at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.messaging.converter.MessageConversionException: No converter found from actual payload type 'ca.bc.gov.open.crdp.process.models.ScannerPub' to expected payload type 'ca.bc.gov.open.crdp.process.models.ScannerPub'
    at org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.convertPayload(MessageMethodArgumentResolver.java:147)
    at org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.resolveArgument(MessageMethodArgumentResolver.java:94)
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
    at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:75)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:277)
    ... 17 common frames omitted
2023-10-27 21:51:21,001 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-2] WARN  org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler$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={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=true, receivedExchange=crdp-exchange, receivedRoutingKey=SCANNER, deliveryTag=6, consumerTag=amq.ctag-p37Eiyz_B5hZDTrhkBlk6Q, consumerQueue=scanner-queue])
2023-10-27 21:51:21,002 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#0-2] ERROR org.springframework.amqp.rabbit.listener.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:147)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeErrorHandler(AbstractMessageListenerContainer.java:1453)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.handleListenerException(AbstractMessageListenerContainer.java:1751)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListenerAndHandleException(AbstractMessageListenerContainer.java:1527)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.lambda$executeListener$8(AbstractMessageListenerContainer.java:1484)
    at io.micrometer.observation.Observation.lambda$observe$0(Observation.java:493)
    at io.micrometer.observation.Observation.observeWithContext(Observation.java:603)
    at io.micrometer.observation.Observation.observe(Observation.java:492)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:1484)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:994)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:941)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.mainLoop(SimpleMessageListenerContainer.java:1323)
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1225)
    at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void ca.bc.gov.open.crdp.process.transformer.services.ConsumerService.receiveScannerPubMessage(org.springframework.messaging.Message<ca.bc.gov.open.crdp.process.models.ScannerPub>)]
Bean [ca.bc.gov.open.crdp.process.transformer.services.ConsumerService@74e89b46]
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:281)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandlerAndProcessResult(MessagingMessageListenerAdapter.java:224)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.onMessage(MessagingMessageListenerAdapter.java:149)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:1663)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.actualInvokeListener(AbstractMessageListenerContainer.java:1582)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:1570)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:1561)
    at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListenerAndHandleException(AbstractMessageListenerContainer.java:1506)
    ... 10 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: No converter found from actual payload type 'ca.bc.gov.open.crdp.process.models.ScannerPub' to expected payload type 'ca.bc.gov.open.crdp.process.models.ScannerPub'
    at org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.convertPayload(MessageMethodArgumentResolver.java:147)
    at org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.resolveArgument(MessageMethodArgumentResolver.java:94)
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:118)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:147)
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:115)
    at org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:75)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:277)
    ... 17 common frames omitted

我进入了包org.springframework.messaging.handler.annotation.support（spring-core6.0.11）中的ObjectresolveArgument（MethodParameter参数，Message消息）。在第 45 行（请参阅随附的屏幕截图），targetPayloadType.isInstance(payload) 应该返回 true，与旧版本 Java 11/Spring Boot 2.7.9 中相同，但现在在 Java 17/Spring Boot 3.1 中返回 false。 2 并触发类型转换，引发异常。

如果 targetPayloadType 是类类型“ca.bc.gov.open.crdp.process.models.ScannerPub”并且有效负载是类 ScannerPub 的对象，则 targetPayloadType.isInstance(payload) 应该为 true 吗？

谢谢你的帮助！

Answer 1

Chasca，旧版本在 RabbitMQ 队列中发送 Java 对象，它运行良好，但升级后就不行了。因此，为了解决这个问题，我只是添加了 JSON 序列化/反序列化。问题解决了。

@Configuration

@ComponentScan 公共类 RabbitMqConfig {

@Bean
public Jackson2JsonMessageConverter converter(){
    return new Jackson2JsonMessageConverter();
}
@Bean
public AmqpTemplate amqpTemplate(ConnectionFactory connectionFactory){
    RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    rabbitTemplate.setMessageConverter(converter());
    return rabbitTemplate;
}

}