Skip to content
优先级、TTL 与惰性队列

优先级、TTL 与惰性队列

三种队列级高级特性:优先级队列(仅 Classic Queue 支持)、消息 TTL 与队列 TTL 的两层设置及注意事项、惰性队列(Lazy Queue)与 Quorum Queue 的取舍。


优先级队列(Priority Queue)

概念

优先级队列允许高优先级的消息先于低优先级消息被消费。队列声明时必须设置最大优先级,声明后不可更改(需要删除重建)。

重要限制:优先级队列仅 Classic Queue 支持,Quorum Queue 和 Stream 均不支持优先级

声明优先级队列

// 声明支持优先级的经典队列
queueInfo, err := conn.Management().DeclareQueue(ctx,
    &rmq.ClassicQueueSpecification{
        Name:       "priority.tasks",
        IsDurable:  true,
        IsExclusive: false,
        IsAutoDelete: false,
        // 通过参数设置最大优先级(0-255,建议 1-10)
        // Arguments: map[string]interface{}{
        //     "x-max-priority": 10,
        // },
    })

注意x-max-priority 的具体设置方式取决于 AMQP 1.0 Go 客户端的 ClassicQueueSpecification 是否暴露了 Arguments 字段。如果库未直接支持,建议使用 RabbitMQ Policy 配置。

# 通过 Policy 配置优先级
rabbitmqctl set_policy PRIORITY-POLICY "^priority\." \
  '{"x-max-priority":10}' --apply-to queues

发布带优先级的消息

publisher, err := conn.NewPublisher(ctx,
    &rmq.QueueAddress{Queue: "priority.tasks"}, nil)
if err != nil {
    log.Fatalf("创建发布者失败: %v", err)
}
defer publisher.Close(context.Background())

// 发送高优先级消息
highPriorityMsg := rmq.NewMessage([]byte("紧急任务:系统告警处理"))
highPriorityMsg.SetPriority(9) // 0-10,越大越优先
// 或者通过注解设置
// highPriorityMsg.SetMessageAnnotation("x-priority", 9)

result, err := publisher.Publish(ctx, highPriorityMsg)
if err != nil {
    log.Printf("发布失败: %v", err)
}

// 发送普通优先级消息
normalMsg := rmq.NewMessage([]byte("普通任务:日志归档"))
normalMsg.SetPriority(3)
publisher.Publish(ctx, normalMsg)

// 发送低优先级消息
lowMsg := rmq.NewMessage([]byte("低优先级:数据报表生成"))
lowMsg.SetPriority(1)
publisher.Publish(ctx, lowMsg)

消费者(无需特殊处理)

// 消费者代码与普通队列完全相同,Broker 自动按优先级出队
consumer, err := conn.NewConsumer(ctx, "priority.tasks",
    &rmq.ConsumerOptions{InitialCredits: 1})
if err != nil {
    log.Fatalf("创建消费者失败: %v", err)
}
defer consumer.Close(context.Background())

for {
    delivery, _ := consumer.Receive(ctx)
    msg := delivery.Message()
    priority := msg.Priority()

    log.Printf("处理优先级=%d 的任务: %s", priority, string(msg.Data()[0]))
    processTask(msg)

    delivery.Accept(ctx)
}

优先级队列表现与开销

优先级范围 CPU/内存开销 适用场景
1-5 简单的两级分类(普通/加急)
1-10 多级任务调度
1-255 复杂优先级体系(不推荐,开销过大)

建议:将 x-max-priority 设为 1-10 即可满足绝大多数场景。


消息 TTL 与队列 TTL

概念

TTL(Time-To-Live)用于控制消息在队列中的存活时间。过期的消息会被自动删除或转发到死信队列。

RabbitMQ 支持两层 TTL

层级 作用范围 设置方式
队列级 TTL 队列中所有消息使用统一过期时间 声明队列时指定
消息级 TTL 每条消息可以有不同的过期时间 发布消息时指定

当两者同时设置时,取较短的那个时间

队列级别 TTL

// 声明带 TTL 的队列——所有消息统一 30 秒后过期
queueInfo, err := conn.Management().DeclareQueue(ctx,
    &rmq.QuorumQueueSpecification{
        Name: "order.timeout.queue",
        // 队列级 TTL(毫秒)
        // MessageTTL: 30000, // 具体字段名取决于库版本
    })

// 配合死信队列使用——过期消息自动转发
// DeadLetterExchange:   "dlx.order",
// DeadLetterRoutingKey: "order.timeout",

消息级别 TTL

publisher, _ := conn.NewPublisher(ctx,
    &rmq.QueueAddress{Queue: "order.timeout.queue"}, nil)
defer publisher.Close(context.Background())

// 消息 1:10 秒后过期
msg1 := rmq.NewMessage([]byte("普通订单"))
msg1.SetTTL(10000) // 毫秒
publisher.Publish(ctx, msg1)

// 消息 2:60 秒后过期
msg2 := rmq.NewMessage([]byte("VIP 订单,更长处理时间"))
msg2.SetTTL(60000) // 毫秒
publisher.Publish(ctx, msg2)

// 消息 3:通过注解设置 TTL
msg3 := rmq.NewMessage([]byte("通过注解设置"))
msg3.SetMessageAnnotation("x-message-ttl", 15000)
publisher.Publish(ctx, msg3)

TTL 关键注意事项

注意点 说明
TTL 单位是毫秒 设 30 秒应写 30000,误写 30 消息立即过期
消息级 TTL 优于队列级 两者同时存在时取较短值
过期 ≠ 立即删除 消息可能在队列中"等"到被投递时才检查过期
无 DLX 时过期消息直接丢弃 必须配合 DLX 才能保留过期消息

惰性队列(Lazy Queue)

概念

惰性队列尽可能早地将消息写入磁盘,仅在内存中保留必要的元数据。适用于消息积压严重但内存有限的场景。

// 声明惰性队列
queueInfo, err := conn.Management().DeclareQueue(ctx,
    &rmq.ClassicQueueSpecification{
        Name:       "lazy.analytics.queue",
        IsDurable:  true,
        IsExclusive: false,
        // 设置为惰性模式
        // Arguments: map[string]interface{}{
        //     "x-queue-mode": "lazy",
        // },
    })
特性 惰性队列 普通经典队列
内存占用 极低(仅存元数据) 尽量缓存消息在内存
磁盘 I/O 高(每次都读写磁盘) 低(优先内存操作)
适用场景 消息积压大、处理不稳定 消息快速流转、内存充足
吞吐量 略低于普通队列 正常

建议:如果使用 Quorum Queue,无需设置此模式——Quorum Queue 已经做了类似的磁盘优先策略。