RabbitMQ 事务机制与消息限流实战详解

2. 配置 RabbitTemplate 和事务管理器 需要手动创建 Bean 来启用通道事务：

@Configuration
public class RabbitTemplateConfig {
    @Bean("transRabbitTemplate")
    public RabbitTemplate transRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setChannelTransacted(true); // 开启事务模式
        return rabbitTemplate;
    }

    @Bean
    public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }
}

3. 队列声明 确保队列持久化以配合事务场景：

@Configuration
public class RabbitMQConfig {
    @Bean("transQueue")
    public Queue transQueue() {
        return QueueBuilder.durable(Constants.TRANS_QUEUE).build();
    }
}

2. 消息限流

消息限流（Flow Control）是 RabbitMQ 防止生产者发送速度超过消费者处理能力的关键机制，能有效避免消息积压和系统崩溃。

核心原理

通过设置 prefetch 数量限制每个消费者未确认的最大消息数。当达到上限时，消费者暂停接收新消息，直到处理完部分旧消息并发送 ACK。

消费者配置

我们需要将监听器配置为手动确认模式（Manual Ack），并在代码中控制确认逻辑。

@Component
@Slf4j
public class QosListener {
    @RabbitListener(queues = Constants.QOS_QUEUE)
    public void handMessage(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            log.info("接收到消息:{},deliveryTag:{}", 
                new String(message.getBody(), StandardCharsets.UTF_8), deliveryTag);
            log.info("处理成功");
            // 处理成功后手动确认
            // channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            // 处理失败，拒绝消息并重新入队
            channel.basicNack(deliveryTag, true, true);
        }
    }
}

全局参数设置

在 application.yml 中配置监听器的简单模式参数：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 开启手动确认
        prefetch: 5              # 限制每批最多预取 5 条消息