Spring AMQP中TestRabbitTemplate的消息转换器异常问题

在消息代理（Broker）未运行时，使用TestRabbitTemplate对Spring AMQP的RabbitListener做集成测试。已为TestRabbitTemplate配置Jackson2JsonMessageConverter，且使用了Jackson注解的CustomMessage对象，但发送消息到交换机时触发ListenerExecutionFailedException，核心原因是无法将byte[]类型转换为com.murex.em.demo.springamqpdemo.message.CustomMessage类型。

异常信息 #

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [[B] to [com.murex.em.demo.springamqpdemo.message.CustomMessage] for GenericMessage [payload=byte[27], headers={amqp_contentEncoding=UTF-8, amqp_contentLength=27, amqp_replyTo=testRabbitTemplateReplyTo, id=fb0352d2-2abb-ae06-9b2e-03da1fc19e43, amqp_lastInBatch=false, contentType=application/json, __TypeId__=com.murex.em.demo.springamqpdemo.message.CustomMessage, timestamp=1676625212607}], failedMessage=GenericMessage [payload=byte[27], headers={amqp_contentEncoding=UTF-8, amqp_contentLength=27, amqp_replyTo=testRabbitTemplateReplyTo, id=fb0352d2-2abb-ae06-9b2e-03da1fc19e43, amqp_lastInBatch=false, contentType=application/json, __TypeId__=com.murex.em.demo.springamqpdemo.message.CustomMessage, timestamp=1676625212607}]
    at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145)
    at org.springframework.amqp.rabbit.annotation.RabbitListenerAnnotationBeanPostProcessor$OptionalEmptyAwarePayloadArgumentResolver.resolveArgument(RabbitListenerAnnotationBeanPostProcessor.java:1053)
    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:77)
    at org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:263)
    ... 78 more

相关代码片段 #

CustomMessage 实体类 #

@Jacksonized
@Builder
@Value
public class CustomMessage {

    String id;

    String name;

}

消息消费者类 #

@Component
public class InwardMessageConsumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "inward.messageQueue", durable = "true"),
            exchange = @Exchange(value = "inward", type = "topic"),
            key = "inwardRoutingKey")
    )
    public String processMessage(CustomMessage customMessage) {
        return customMessage.getName();

    }

}

测试类 #

@SpringJUnitConfig
@SpringBootTest
class BrokerNotRunningIT {

    @Autowired
    private TestRabbitTemplate template;

    @Test
    public void testSendAndReceive() {
        CustomMessage customMessage = CustomMessage.builder()
                .id("123")
                .name("name1")
                .build();

        assertThat(template.convertSendAndReceive("inward", "inward.messageQueue", customMessage)).isEqualTo("name1");
    }

    @TestConfiguration
    public static class RabbitTestConfiguration {

        @Bean
        public TestRabbitTemplate testRabbitTemplate(
                ConnectionFactory mockConnectionFactory,
                Jackson2JsonMessageConverter jackson2JsonMessageConverter
        ) {
            TestRabbitTemplate testRabbitTemplate = new TestRabbitTemplate(mockConnectionFactory);
            testRabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
            return testRabbitTemplate;
        }

        @Bean
        public ConnectionFactory mockConnectionFactory() throws IOException {
            ConnectionFactory factory = mock(ConnectionFactory.class);
            Connection connection = mock(Connection.class);
            Channel channel = mock(Channel.class);
            AMQP.Queue.DeclareOk declareOk = mock(AMQP.Queue.DeclareOk.class);
            willReturn(connection).given(factory).createConnection();
            willReturn(channel).given(connection).createChannel(anyBoolean());
            given(channel.isOpen()).willReturn(true);
            given(channel.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), anyMap()))
                    .willReturn(declareOk);
            return factory;
        }

        @Bean
        public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            return factory;
        }

        @Bean
        public SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
            return new SimpleMessageListenerContainer(connectionFactory);
        }
    }
}

解决方案 #

问题核心 #

仅为发送端的TestRabbitTemplate配置了JSON消息转换器，但消费消息的容器工厂未配置同款转换器。发送时消息被序列化为JSON字节数组，消费者收到后用默认转换器无法完成反序列化。

修复步骤 #

1. 为消费者容器工厂添加消息转换器
   修改测试配置中的SimpleRabbitListenerContainerFactory Bean，注入并设置Jackson2JsonMessageConverter：

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
        ConnectionFactory connectionFactory,
        Jackson2JsonMessageConverter jackson2JsonMessageConverter
) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    // 为消费者配置JSON消息转换器
    factory.setMessageConverter(jackson2JsonMessageConverter);
    return factory;
}

2. 确保Jackson消息转换器被声明为Bean
   如果项目未自动配置，需在测试配置中添加：

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

3. 修正发送消息的路由键参数
   测试方法中convertSendAndReceive的第二个参数应为路由键inwardRoutingKey，而非队列名称：

assertThat(template.convertSendAndReceive("inward", "inwardRoutingKey", customMessage)).isEqualTo("name1");

完成上述修改后，消费者容器会使用JSON转换器将字节数组反序列化为CustomMessage对象，监听方法即可正常执行。

内容的提问来源于stack exchange，提问作者Poet