两个生产者之后：给无锁环形队列加同步的三条路

type MPSC struct {
    buf       []entry
    head      int64 // 认领边界：生产者 CAS 抢
    committed int64 // 发布边界：轮到的生产者推进
    tail      int64 // 消费者独占写
}

func (q *MPSC) Push(e entry) bool {
    for {
        head := atomic.LoadInt64(&q.head)
        if head-atomic.LoadInt64(&q.tail) >= int64(len(q.buf)) {
            return false // 满：认领前退出，无副作用
        }
        // 认领：只有一个生产者能打成功，槽位号 = head
        if !atomic.CompareAndSwapInt64(&q.head, head, head+1) {
            continue // 被抢了，重读重试
        }
        q.buf[head&(int64(len(q.buf))-1)] = e // 槽位已归我，放心写
        for atomic.LoadInt64(&q.committed) != head {
            runtime.Gosched() // 排队交卷：等前面的生产者先发布
        }
        atomic.StoreInt64(&q.committed, head+1) // release：数据先落地，边界后推进
        return true
    }
}

func (q *MPSC) Pop() (entry, bool) {
    tail := q.tail
    if tail == atomic.LoadInt64(&q.committed) { // acquire：与 release 配对
        return entry{}, false // 空
    }
    e := q.buf[tail&(int64(len(q.buf))-1)]
    atomic.StoreInt64(&q.tail, tail+1) // 释放槽位
    return e, true
}

type MPSCMutex struct {
    mu   sync.Mutex // 只罩生产者
    buf  []entry
    head int64 // 生产者写（持锁），消费者读
    tail int64 // 消费者独占写
}

func (q *MPSCMutex) Push(e entry) bool {
    q.mu.Lock()
    defer q.mu.Unlock()
    if q.head-atomic.LoadInt64(&q.tail) >= int64(len(q.buf)) {
        return false // 满
    }
    q.buf[q.head&(int64(len(q.buf))-1)] = e
    atomic.StoreInt64(&q.head, q.head+1) // 锁保写权，atomic 保发布
    return true
}

func (q *MPSCMutex) Pop() (entry, bool) {
    // 与 SPSC 版完全一致：tail 仍是唯一写者，消费者无锁
    tail := q.tail
    if tail == atomic.LoadInt64(&q.head) {
        return entry{}, false
    }
    e := q.buf[tail&(int64(len(q.buf))-1)]
    atomic.StoreInt64(&q.tail, tail+1)
    return e, true
}

type cell struct {
    seq  int64 // 状态机：== pos 可写，== pos+1 可读
    data entry
    _    [56]byte // 补齐到独立 64B 缓存行（按 entry 大小调整）
}

type MPMC struct {
    buf    []cell // 容量为 2 的幂
    mask   int64
    enqPos int64 // 生产者 CAS 认领
    deqPos int64 // 消费者 CAS 认领
}

func NewMPMC(capacity int64) *MPMC {
    q := &MPMC{buf: make([]cell, capacity), mask: capacity - 1}
    for i := range q.buf {
        q.buf[i].seq = int64(i) // 初始：槽 i 属于第一圈的 pos=i
    }
    return q
}

func (q *MPMC) Push(e entry) bool {
    pos := atomic.LoadInt64(&q.enqPos)
    for {
        c := &q.buf[pos&q.mask]
        switch seq := atomic.LoadInt64(&c.seq); {
        case seq == pos: // 轮到我：本圈此槽可写
            if atomic.CompareAndSwapInt64(&q.enqPos, pos, pos+1) {
                c.data = e                       // 先写数据
                atomic.StoreInt64(&c.seq, pos+1) // 后发布（release）
                return true
            }
        case seq < pos: // 上一圈的数据还没被消费 → 满
            return false
        default: // 别人先认领了，刷新重试
            pos = atomic.LoadInt64(&q.enqPos)
        }
    }
}

func (q *MPMC) Pop() (entry, bool) {
    pos := atomic.LoadInt64(&q.deqPos)
    for {
        c := &q.buf[pos&q.mask]
        switch seq := atomic.LoadInt64(&c.seq); {
        case seq == pos+1: // 本圈数据就位
            if atomic.CompareAndSwapInt64(&q.deqPos, pos, pos+1) {
                e := c.data                             // 先读数据
                atomic.StoreInt64(&c.seq, pos+q.mask+1) // 后释放（= pos+cap，下圈可写）
                return e, true
            }
        case seq < pos+1: // 本圈还没发布 → 空
            return entry{}, false
        default: // 别人先认领了，刷新重试
            pos = atomic.LoadInt64(&q.deqPos)
        }
    }
}

type MergedQ struct {
    inbox chan entry // 任意 goroutine 都能投递
    spsc  SPSC       // 第一篇的实现，原样复用
}

// 唯一的转发者：SPSC 的唯一生产者
func (m *MergedQ) Run() {
    for e := range m.inbox {
        for !m.spsc.Push(e) {
            runtime.Gosched() // 满：等消费者腾地方
        }
    }
}