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

队列定义保持持久化即可：

@Configuration
public class RabbitMQConfig {

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

2. 消息限流 (QoS)

消息限流是防止生产者发送速度超过消费者处理能力的关键保护机制，能有效避免消息积压导致系统崩溃。

核心原理

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

代码实现

生产者端正常发送即可：

@RestController
@RequestMapping("/producer")
public class ProducerController {

    @Resource(name = "rabbitTemplate")
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("/qos")
    public String qos() {
        for (int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend(Constants.QOS_EXCHANGE, "qos", "qos test:" + i);
        }
        return "发送成功";
    }
}

消费者端需要手动管理 ACK/NACK，并配置监听器的 acknowledge-mode 为 manual。

@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) {
            // 拒绝消息，true 表示重入队列
            channel.basicNack(deliveryTag, true, true);
        }
    }
}

配置文件需指定 prefetch 数量，这里设置为 5：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 开启手动确认
        prefetch: 5              # 限制未确认最大消息数为 5

交换器与队列绑定关系配置：

@Configuration
public class RabbitMQConfig {

    @Bean("qosQueue")
    public Queue qosQueue() {
        return QueueBuilder.durable(Constants.QOS_QUEUE).build();
    }

    @Bean("qosExchange")
    public DirectExchange qosExchange() {
        return ExchangeBuilder.directExchange(Constants.QOS_EXCHANGE).build();
    }

    @Bean("qosBinding")
    public Binding qosBinding(@Qualifier("qosExchange") DirectExchange directExchange,
                              @Qualifier("qosQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(directExchange).with("qos");
    }
}