使用DirectMessageListenerContainer压缩解压Spring RabbitMQ消息遇问题

看起来你在切换到DirectMessageListenerContainer并添加GZip压缩/解压的MessagePostProcessor后踩了几个坑：断点完全没触发、@RabbitListener收不到消息，甚至容器好像还在沿用旧的SimpleMessageListenerContainer。我帮你一步步拆解排查，找到问题根源：

1. 先确认容器是否真的切换成功 #

这是最基础的前提——如果容器没换成DirectMessageListenerContainer，你的新配置自然不会生效。

• Java配置检查：确保配置类里返回的是DirectMessageListenerContainer而非旧容器：

  @Bean
  public DirectMessageListenerContainer directMessageListenerContainer(ConnectionFactory connectionFactory) {
      DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
      container.setQueueNames("your-target-queue");
      container.setConcurrentConsumers(2);
      // 后续添加处理器的逻辑要写在这里
      return container;
  }
  

• XML配置检查：如果用XML，要给listener容器指定container-type="direct"：

  <rabbit:listener-container connection-factory="connectionFactory" container-type="direct">
      <rabbit:listener queues="your-target-queue" ref="yourMessageListenerBean"/>
  </rabbit:listener-container>
  

• 日志验证：启动项目时搜DirectMessageListenerContainer，确认能看到该容器初始化的日志，而不是SimpleMessageListenerContainer的启动日志。

2. 纠正MessagePostProcessor的注册方式 #

DirectMessageListenerContainer和旧容器的处理器注册逻辑不一样，用旧方法绑定肯定失效：

消费端（解压）的正确绑定 #

要把解压处理器注册到容器的afterReceivePostProcessors里——这是消息被接收后、传给监听器前的处理环节：

@Bean
public DirectMessageListenerContainer directMessageListenerContainer(ConnectionFactory connectionFactory,
                                                                   MessagePostProcessor gUnzipPostProcessor) {
    DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
    container.setQueueNames("your-target-queue");
    // 关键：把解压处理器绑定到接收后处理链
    container.setAfterReceivePostProcessors(gUnzipPostProcessor);
    return container;
}

// 解压处理器实现（这里加断点，后续验证是否触发）
@Bean
public MessagePostProcessor gUnzipPostProcessor() {
    return message -> {
        // 先判断消息是否是压缩过的（建议生产端加Content-Encoding头标记）
        if ("gzip".equals(message.getMessageProperties().getHeaders().get("Content-Encoding"))) {
            byte[] compressedBody = message.getBody();
            try (GZIPInputStream gzis = new GZIPInputStream(new ByteArrayInputStream(compressedBody))) {
                byte[] decompressedBody = StreamUtils.copyToByteArray(gzis);
                return MessageBuilder.fromMessage(message)
                        .setBody(decompressedBody)
                        .removeHeader("Content-Encoding")
                        .build();
            }
        }
        return message;
    };
}

生产端（压缩）的正确绑定 #

发送时压缩要把处理器绑定到RabbitTemplate的beforePublishPostProcessors，或者在发送时手动传入：

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessagePostProcessor gZipPostProcessor) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    // 全局绑定发送前压缩处理器
    template.setBeforePublishPostProcessors(gZipPostProcessor);
    return template;
}

@Bean
public MessagePostProcessor gZipPostProcessor() {
    return message -> {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        try (GZIPOutputStream gzos = new GZIPOutputStream(bos)) {
            gzos.write(message.getBody());
        }
        byte[] compressedBody = bos.toByteArray();
        return MessageBuilder.fromMessage(message)
                .setBody(compressedBody)
                // 加标记头，方便消费端判断
                .setHeader("Content-Encoding", "gzip")
                .build();
    };
}

3. 修复@RabbitListener不生效的问题 #

如果用注解式监听器，还要确保容器和注解正确关联：

• 配置类必须加@EnableRabbit，开启注解支持。
• 推荐用RabbitListenerContainerFactory统一管理容器，这样@RabbitListener可以直接指定使用你的自定义容器：

  @Bean
  public RabbitListenerContainerFactory<DirectMessageListenerContainer> directContainerFactory(
          ConnectionFactory connectionFactory, MessagePostProcessor gUnzipPostProcessor) {
      DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
      factory.setConnectionFactory(connectionFactory);
      factory.setAfterReceivePostProcessors(gUnzipPostProcessor);
      factory.setConcurrentConsumers(2);
      return factory;
  }
  

  然后在注解上指定这个工厂：

  @RabbitListener(queues = "your-target-queue", containerFactory = "directContainerFactory")
  public void handleMessage(YourPayloadDto payload) {
      // 处理业务逻辑
  }
  

4. 断点未触发的快速排查 #

如果处理器里的断点没触发，说明处理器根本没被调用：

• 先打印容器的处理器数量，确认是否真的绑定成功：

  @Autowired
  private DirectMessageListenerContainer container;
  
  @PostConstruct
  public void checkProcessors() {
      System.out.println("已绑定的接收后处理器数量：" + container.getAfterReceivePostProcessors().length);
  }
  

• 去RabbitMQ管理控制台看消息：如果是生产端问题，压缩后的消息body应该比原消息小；如果是消费端问题，确认消息确实被投递到了监听的队列。

5. 注意DirectMessageListenerContainer的特性 #

这个容器和旧容器的工作机制不同，别踩特性坑：

• 它是每个消费者线程直接从RabbitMQ拉取消息，没有预取缓存，所以并发数要根据业务吞吐量合理设置。
• 确保你的MessagePostProcessor是线程安全的——多个消费者线程会共用同一个处理器实例。

最后建议先简化配置：先让DirectMessageListenerContainer不带处理器的情况下正常接收消息，确认@RabbitListener能正常工作，再逐步添加压缩/解压逻辑，这样更容易定位问题。

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