如何保证 MQ 消息的顺序消费？

做法

Topic 只设 1 个 partition（或 1 个 queue），只起 1 个 consumer 实例，且该实例单线程顺序消费。Broker 保证分区内有序，单分区 + 单线程 → 全局有序。

发送端：让同 key 进同分区

Kafka 的 HashPartitioner 默认按 msg.Key 哈希选分区——只要把业务键设为 Key，同 key 自然同分区。下面显式写出，避免歧义：

func produce(p sarama.SyncProducer, orderID string, body []byte) {
    msg := &sarama.ProducerMessage{
        Topic: "order",
        Key:   sarama.StringEncoder(orderID),  // 关键：业务键当 Key
        Value: sarama.ByteEncoder(body),
    }
    p.SendMessage(msg)  // HashPartitioner 按 Key 哈希 → 同 orderID 同分区
}

若用自定义分区逻辑（如要控制分区数），实现 sarama.Partitioner：

func (p *keyPartitioner) Partition(msg *sarama.ProducerMessage, n int32) (int32, error) {
    h := fnv32(string(msg.Key.([]byte)))  // 稳定哈希
    return int32(h % uint32(n)), nil
}

消费端：每个分区单线程串行

sarama 的 ConsumerGroup 对每个被 claim 的分区起一个 goroutine 跑 ConsumeClaim。所以「在该循环里顺序处理、不另起 goroutine 并发」就保证分区内有序：

func (h *orderHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {  // 本分区串行
        h.process(msg)              // 务必同步处理完再取下一条
        sess.MarkMessage(msg, "")   // 处理成功才提交 offset
    }
    return nil
}

发送：用 MessageQueueSelector 把同 key 钉进同一队列

p, _ := rocketmq.NewProducer(
    producer.WithNameServer([]string{"127.0.0.1:9876"}),
)
// 选队列：orderID 哈希 % 队列数，确保同订单同队列
selector := func(qs []*primitive.MessageQueue, msg *primitive.Message, arg interface{}) *primitive.MessageQueue {
    orderID := arg.(string)
    idx := fnv32(orderID) % len(qs)
    return qs[idx]
}
err := p.SendMessageOrderly(context.Background(),
    []*primitive.Message{primitive.NewMessage("order_topic", body)},
    selector, orderID)  // 第三个参数是 selector 的 arg

SendMessageOrderly 会「先拿队列锁再发」，保证同队列的消息严格按发送顺序落盘。

消费：注册 MessageListenerOrderly

RocketMQ 客户端在消费端对每个 MessageQueue 加锁，同一队列的消息被单线程串行回调 ConsumeOrderly，天然有序：

c, _ := rocketmq.NewPushConsumer(consumer.WithNameServer([]string{"127.0.0.1:9876"}))
c.RegisterMessageListener(func(msgs []*primitive.MessageExt, ce consumer.ConsumeOrderlyContext) consumer.ConsumeResult {
    for _, m := range msgs {  // 同一队列串行到达，已保序
        if err := process(m); err != nil {
            ce.SetSuspendCurrentQueueTimeMillis(1000)  // 失败：挂起当前队列稍后重试（不乱序）
            return consumer.SuspendCurrentQueueAMoment
        }
    }
    return consumer.ConsumeRetryLater  // 成功
})

完整 Go 实现

type Dispatcher struct {
    buckets []chan *Msg
    stop    chan struct{}
}
func NewDispatcher(n int, process func(*Msg)) *Dispatcher {
    d := &Dispatcher{buckets: make([]chan *Msg, n), stop: make(chan struct{})}
    for i := range d.buckets {
        d.buckets[i] = make(chan *Msg, 1024)  // 缓冲防背压阻塞拉取
        go d.worker(i, process)             // 每个桶一个固定 goroutine
    }
    return d
}
func (d *Dispatcher) Dispatch(m *Msg) {
    idx := fnv32(m.Key) % len(d.buckets)  // 同 key 恒进同桶
    d.buckets[idx] <- m
}
func (d *Dispatcher) worker(i int, process func(*Msg)) {
    for m := range d.buckets[i] {  // 单 goroutine：天然有序
        process(m)
    }
}

完整 Go 实现（按 key 维护水位线）

type Msg struct { Key string; Seq int64; Payload []byte }
type Tracker struct {
    mu   sync.Mutex
    last map[string]int64
    hold map[string]map[int64]*Msg
}
// Feed 返回「可以安全按顺序处理的消息序列」
func (t *Tracker) Feed(m *Msg) []*Msg {
    t.mu.Lock(); defer t.mu.Unlock()
    last, ok := t.last[m.Key]
    if ok && m.Seq <= last { return nil }      // 重复/已处理 → 丢弃（幂等）
    if !ok || m.Seq == last+1 {            // 正好续上
        out := []*Msg{m}; t.last[m.Key] = m.Seq
        for {                                 // 顺手 flush 连续暂存
            n, has := t.hold[m.Key][t.last[m.Key]+1]
            if !has { break }
            delete(t.hold[m.Key], t.last[m.Key]+1)
            t.last[m.Key]++; out = append(out, n)
        }
        return out
    }
    if t.hold[m.Key] == nil { t.hold[m.Key] = map[int64]*Msg{} }
    t.hold[m.Key][m.Seq] = m                    // 有缺口 → 暂存
    return nil
}

seq 可由生产者用「key + 自增计数器」生成，或复用数据库行的版本号 / 事件时间戳（需保证同 key 单调）。

去重表：以 (key, seq) 或 messageID 唯一约束

// Redis：SETNX 一条，存在即重复
func dedup(rdb *redis.Client, key string, seq int64) (bool, error) {
    ok, err := rdb.SetNX(context.Background(),
        "dedup:"+key+":"+strconv.FormatInt(seq, 10), 1,
        24*time.Hour).Result()
    return ok, err  // ok=false → 已处理过，直接跳过
}

// 或数据库：业务表加唯一索引 (biz_key, seq)，插入冲突即重复
// ALTER TABLE order_event ADD UNIQUE KEY uk_key_seq (biz_key, seq);

消费逻辑：if !dedup(key,seq) { return } 再执行业务。这样「顺序乱了重投」和「正常重投」都被同一道闸拦住。