@RabbitListener containerFactory 的奇怪行为

Question

RabbitMQ 消息到达时带有类型信息：

headers:    
__ContentTypeId__:  java.lang.Object
__TypeId__: java.util.ArrayList

我在代码中有这 2 个容器工厂 bean - 默认一个使用转换器，另一个不使用：

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

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

场景 1. 一个听众。

@RabbitListener(queues = "#{@myQueueBean}")
public void persistMessage(Message message) {
    .....
}

当我检查到达监听器的内容时​​，message.payload 是 ArrayList，其中包含 HashMaps。这看起来不错，似乎使用 rabbitListenerContainerFactory 作为默认值。

场景2.我改用不带转换器的containerFactory：

@RabbitListener(queues = "#{@myQueueBean}", containerFactory = "rawContainerFactory")
public void persistMessage(Message message) {
    .....
}

当我检查到达的内容时，message.payload 现在是带有消息内容的 byte[]。再次看起来不错。

场景3.我改为显式使用默认的containerFactory：

@RabbitListener(queues = "#{@myQueueBean}", containerFactory = "rabbitListenerContainerFactory")
public void persistMessage(Message message) {
    .....
}

当我检查到达的内容时，message.payload 又是带有消息内容的 byte[]。这听起来出乎意料，我希望 ArrayList 就像场景 1 中一样。

现在我在第一个 bean 中注释掉 messageConverter：

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

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

场景 4. 再次监听器

@RabbitListener(queues = "#{@myQueueBean}")
public void persistMessage(Message message) {
    .....
}

当我检查到达的内容时，我希望 message.payload 为 byte[]（使用默认的 containerfactory，没有像场景 2 这样的转换器），但实际上我得到的是一个逗号分隔数字的字符串。这听起来不对。

场景5.我改为显式使用默认containerFactory，不带转换器：

@RabbitListener(queues = "#{@myQueueBean}", containerFactory = "rabbitListenerContainerFactory")
public void persistMessage(Message message) {
    .....
}

当我检查到达的内容时，message.payload 现在是带有消息内容的 byte[]。这就像场景 2，看起来还不错。

场景 6. 使用“rawContainerFactory”，结果是现在逻辑上预期的 byte[] 数组。

场景 7. 使用“rawContainerFactory”并将 Jackson 转换器添加到其中。结果是 ArrayList。情景 3 结果出乎意料。

上面的结果看起来非常不一致。 “rabbitListenerContainerFactory”有一些魔力，它提示我说 - 忘记它并在显式配置中使用自己的自定义工厂。除非有人设法解释这种魔法。

更新：我无法再重现该问题，这也是它消失的另一个原因。场景 3 和场景 4 的问题都消失了，除了添加一个测试用例以使用 RabbitTemplate 将消息发送到队列之外，我实际上什么也没做（但仍然使用队列中真正的大 json 进行测试）。之前和现在，我都在磁盘上的同一个物理项目代码上使用 Eclipse 和 Idea 进行测试。另外一点，我的示例中的所有侦听器都使用非通用消息类型参数。我这样做是因为在某些场景中添加了任何特定类型（如果我没记错的话，可能是场景 1 或 4）——甚至是 正在将其行为更改为消息甚至无法到达我的听众的程度，并且日志中会出现错误提示“无法将 START_OBJECT 转换为某种类型”。

Answer 1

我无法重现您引用的行为。一切都按我的预期工作。

@SpringBootApplication
public class So47393130Application {

    public static void main(String[] args) {
        SpringApplication.run(So47393130Application.class, args).close();
    }

    @Bean
    public ApplicationRunner runner(RabbitTemplate rabbitTemplate) {
        return args -> {
            rabbitTemplate.convertAndSend("foo", "", new ArrayList<>(Arrays.asList("foo", "bar")));
            Thread.sleep(5_000);
        };
    }

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

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

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

    @Bean
    public Queue foo() {
        return new Queue("foo");
    }

    @Bean
    public Queue bar() {
        return new Queue("bar");
    }

    @Bean
    public Queue baz() {
        return new Queue("baz");
    }

    @Bean
    public FanoutExchange exchange() {
        return new FanoutExchange("foo");
    }

    @Bean
    public Binding fooBinding() {
        return BindingBuilder.bind(foo()).to(exchange());
    }

    @Bean
    public Binding barBinding() {
        return BindingBuilder.bind(bar()).to(exchange());
    }

    @Bean
    public Binding bazBinding() {
        return BindingBuilder.bind(baz()).to(exchange());
    }

    @RabbitListener(queues = "foo")
    public void foo(Message<?> in) {
        System.out.println("foo:" + in.getPayload());
    }

    @RabbitListener(queues = "bar", containerFactory = "rawContainerFactory")
    public void bar(Message<?> in) {
        System.out.println("bar:" + in.getPayload());
    }

    @RabbitListener(queues = "baz", containerFactory = "rabbitListenerContainerFactory")
    public void baz(Message<?> in) {
        System.out.println("baz:" + in.getPayload());
    }

}

也许您可以在侦听器中设置断点并向下查看堆栈以准确查看连接到侦听器适配器的转换器。