RabbitMQ 消息该如何插队

# RabbitMQ 消息该如何插队

## 引言

在消息队列系统中，消息通常按照先进先出（FIFO）的顺序被消费。然而，某些业务场景下（如紧急订单、高优先级任务），需要让特定消息优先被处理。RabbitMQ作为流行的消息中间件，提供了多种实现消息"插队"的机制。本文将深入探讨这些方案的实现原理和适用场景。

## 一、优先级队列（Priority Queue）

### 1. 实现原理
RabbitMQ 3.5.0+ 支持优先级队列，通过`x-max-priority`参数声明队列时指定优先级范围（通常1-10）：

```java
Map<String, Object> args = new HashMap<>();
args.put("x-max-priority", 10);
channel.queueDeclare("priority_queue", true, false, false, args);

发送消息时设置priority属性：

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .priority(5)
    .build();
channel.basicPublish("", "priority_queue", props, message.getBytes());

2. 注意事项

• 优先级高的消息会被提前消费，但不保证严格顺序
• 高优先级消息堆积时，低优先级消息可能被”饿死”
• 需要消费者预取（prefetch）设置为1才能生效

二、多队列优先消费方案

1. 多队列+消费者优先级

graph LR
    Producer -->|高优先级| Queue_High
    Producer -->|普通消息| Queue_Normal
    Consumer --> Queue_High
    Consumer --> Queue_Normal

实现步骤： 1. 创建两个队列：high_priority_queue和normal_queue 2. 消费者优先监听high_priority_queue 3. 使用basicQos控制预取数量

# 先消费高优先级队列
channel.basic_consume(queue='high_priority_queue',
                      on_message_callback=callback)
# 再消费普通队列
channel.basic_consume(queue='normal_queue',
                      on_message_callback=callback)

2. 死信队列升级方案

通过TTL+死信交换器实现延迟降级： 1. 高优先级消息直接进入处理队列 2. 普通消息先进入延迟队列（设置TTL） 3. TTL过期后转入处理队列

三、消息预取控制

通过调整prefetch_count优化消费速度：

// 每次只预取1条消息
channel.basicQos(1);

配合优先级队列使用时，建议： - 高优先级消费者设置较小的prefetch - 低优先级消费者设置较大prefetch

四、业务层解决方案

1. 消息置换方案

消费者处理逻辑：

def callback(ch, method, properties, body):
    if is_high_priority(body):
        # 中断当前处理流程
        ch.basic_nack(method.delivery_tag, requeue=True)
        # 立即获取新消息
        ch.basic_get(queue='priority_queue')
    else:
        process_message(body)

2. 双队列+定时扫描

// 每5秒扫描一次高优先级队列
ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
executor.scheduleAtFixedRate(() -> {
    GetResponse response = channel.basicGet("high_priority_queue", false);
    if (response != null) {
        processMessage(response.getBody());
    }
}, 0, 5, TimeUnit.SECONDS);

五、方案对比

结语

实际生产中，建议优先考虑优先级队列方案。对于金融/医疗等对时效性要求极高的场景，可采用多队列+优先消费的组合方案。无论采用哪种方案，都需要通过监控消息积压情况（如rabbitmqctl list_queues）及时调整策略。

最佳实践：在消息属性中添加timestamp和priority双重标识，便于后续排查消息处理顺序问题。 “`