RabbitMQ 事务机制与消息限流配置实战

配置 RabbitTemplate 和事务管理器

为了启用事务，需要自定义 RabbitTemplate 并设置 channelTransacted 为 true，同时注册事务管理器。

@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);
    }
}

队列配置

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

2. 消息限流

消息限流（Flow Control）是 RabbitMQ 防止生产者发送消息速度超过消费者处理能力，导致消息积压和系统崩溃的保护机制。通过限制消费者未确认的最大消息数，可以有效平衡负载。

发送消息

生产者端正常发送消息即可，重点在于消费端的处理逻辑。

@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 "发送成功";
    }
}

配置消费者手动确认

消费者端需要开启手动确认模式，并在处理完消息后显式调用 basicAck 或 basicNack。这里演示了当发生异常时进行拒绝重投的处理。

@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);
        }
    }
}

全局限流配置

在 application.yml 中配置监听器的确认模式和预取数量（prefetch）。prefetch:5 表示每个消费者最多只持有 5 条未确认的消息。

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 消费者确认机制
        prefetch: 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");
    }
}