RabbitMQ/Spring-AMQP 高级特性：事务机制与消息限流

此外，需要手动配置支持事务的 RabbitTemplate 和事务管理器：

@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）是防止生产者发送速度超过消费者处理能力的重要保护机制。当消费者处理不过来时，通过限流可以避免消息积压甚至系统崩溃。

消费者端配置

我们需要将消费者设置为手动确认模式（Manual Ack），并设置预取数量（Prefetch）。这样消费者每次只获取指定数量的消息进行处理，处理完一条确认一条。

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual # 启用手动 ACK
        prefetch: 5              # 每个消费者最多缓存 5 条未确认消息

消费逻辑示例

在监听器中，我们根据业务处理结果决定是否确认消息。如果处理失败，可以选择重新入队或丢弃。

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

交换器与队列绑定

为了配合限流测试，我们需要正确声明交换器、队列及绑定关系：

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

通过合理配置 prefetch 和手动 ACK，我们可以有效平衡生产者的压力与消费者的负载能力，确保消息处理的稳定性。