六百万 TPS 的单线程：LMAX 架构与 Disruptor 精读

type Sequence struct {
    _ [7]int64 // 左垫
    v int64    // 值
    _ [7]int64 // 右垫
} // 120B：值独占缓存行，邻居碰不到（false sharing，见 07.md 注意事项）

func (s *Sequence) Get() int64           { return atomic.LoadInt64(&s.v) }
func (s *Sequence) Set(v int64)          { atomic.StoreInt64(&s.v, v) }
func (s *Sequence) CAS(old, new int64) bool {
    return atomic.CompareAndSwapInt64(&s.v, old, new)
}

// minSeq：起点和一串序号取最小——"最慢的消费者在哪"只此一个问法
func minSeq(start int64, seqs ...*Sequence) int64 {
    m := start
    for _, s := range seqs {
        if v := s.Get(); v < m { m = v }
    }
    return m
}

type Event struct {
    Raw []byte // 网络原始字节：journaler / replicator / unmarshaler 读
    Req Request // 业务对象：unmarshaler 写、BLP 读
    Out OutputEvent // 输出事件：BLP 写、输出侧 marshal 读
}

type RingBuffer struct {
    slots []Event
    mask  int64
}

func NewRingBuffer(sizePow2 int) *RingBuffer {
    return &RingBuffer{slots: make([]Event, sizePow2), mask: int64(sizePow2 - 1)}
}

func (rb *RingBuffer) slot(seq int64) *Event { return &rb.slots[seq&rb.mask] }

type Sequencer interface {
    Cursor() *Sequence                         // 发布边界，消费者看这里
    AddGating(s ...*Sequence)                  // 登记消费者：满环检查用
    Next() int64                               // 认领槽位（满则自旋等待）
    Publish(seq int64)                         // 数据就位后发布
    HighestPublished(lower, upper int64) int64 // 缺口检查，单生产者直接透传
}

type SingleProducerSequencer struct {
    bufSize int64
    cursor  Sequence   // 发布边界：唯一由 Publish 推进
    gating  []*Sequence
    claim   int64      // 认领边界：只有生产者 goroutine 碰，普通变量即可
    cached  int64      // 缓存的最慢消费者：避免每次跨核读
}

func (p *SingleProducerSequencer) Next() int64 {
    next := p.claim + 1
    if wrap := next - p.bufSize; wrap > p.cached {
        p.cached = minSeq(p.cursor.Get(), p.gating...) // 缓存失效才跨核
        for wrap > p.cached {
            runtime.Gosched()                          // 满：等最慢消费者让位
            p.cached = minSeq(p.cursor.Get(), p.gating...)
        }
    }
    p.claim = next
    return next
}

func (p *SingleProducerSequencer) Publish(seq int64) { p.cursor.Set(seq) }

func (p *SingleProducerSequencer) HighestPublished(lower, upper int64) int64 {
    return upper // 单生产者发布必然连续：cursor 到哪，哪就是连续就绪
}

type MultiProducerSequencer struct {
    bufSize   int64
    mask      int64
    shift     uint     // log2(bufSize)：seq >> shift = 第几圈
    cursor    Sequence // 认领边界：CAS 竞争点
    available []int32  // 每格"第几圈已发布"，-1 = 从未
    gating    []*Sequence
}

// Next：CAS 认领一个槽位（演示版每次重读 gating，生产版加 2.1 的 cached）
func (p *MultiProducerSequencer) Next() int64 {
    for {
        cur := p.cursor.Get()
        next := cur + 1
        if wrap := next - p.bufSize; wrap > minSeq(cur, p.gating...) {
            runtime.Gosched()
            continue
        }
        if p.cursor.CAS(cur, next) {
            return cur // 认领成功，槽位号 = cur
        }
    }
}

// Publish：认领 ≠ 发布。数据写进槽位之后，才标记"本圈就绪"
func (p *MultiProducerSequencer) Publish(seq int64) {
    atomic.StoreInt32(&p.available[seq&p.mask], int32(seq>>p.shift))
}

// HighestPublished：把下游推进界从"认领界"收紧到"连续已发布界"
func (p *MultiProducerSequencer) HighestPublished(lower, upper int64) int64 {
    for seq := lower; seq <= upper; seq++ {
        if atomic.LoadInt32(&p.available[seq&p.mask]) != int32(seq>>p.shift) {
            return seq - 1 // 缺口（认领了没发布）：停在这里，FIFO 不跳过
        }
    }
    return upper
}

type WaitStrategy interface {
    WaitFor(seq int64, cursor *Sequence, deps []*Sequence) int64
}

// 纯自旋：延迟最低、烧核。LMAX 的交易管线用它
type BusySpin struct{}

func (BusySpin) WaitFor(seq int64, cursor *Sequence, deps []*Sequence) int64 {
    for {
        if avail := minSeq(cursor.Get(), deps...); avail >= seq {
            return avail
        }
    }
}

// 自旋 N 次让一步：后台管线用，省核
type Yielding struct{}

func (Yielding) WaitFor(seq int64, cursor *Sequence, deps []*Sequence) int64 {
    for i := 0; ; i++ {
        if avail := minSeq(cursor.Get(), deps...); avail >= seq {
            return avail
        }
        if i&0x3F == 0x3F {
            runtime.Gosched() // 每 64 次 Gosched 一轮
        }
    }
}

type Barrier struct {
    seqr Sequencer
    deps []*Sequence // 前置消费者的序号：链式依赖的"闸门"
    wait WaitStrategy
}

// WaitFor：先等"能追到哪"，再被 availableBuffer 收紧到"连续能追到哪"
func (b *Barrier) WaitFor(seq int64) int64 {
    avail := b.wait.WaitFor(seq, b.seqr.Cursor(), b.deps)
    if avail < seq {
        return avail // 队列空
    }
    return b.seqr.HighestPublished(seq, avail)
}

type BatchProcessor struct {
    seq     Sequence // 自己的消费进度：唯一写者是自己
    barrier *Barrier
    rb      *RingBuffer
    onEvent func(ev *Event, seq int64, endOfBatch bool) // 业务回调：装配时注册
}

func (p *BatchProcessor) Run(stop <-chan struct{}) {
    next := p.seq.Get() + 1
    for {
        select {
        case <-stop:
            return
        default:
        }
        avail := p.barrier.WaitFor(next)
        if avail >= next {
            for i := next; i <= avail; i++ {
                p.onEvent(p.rb.slot(i), i, i == avail) // endOfBatch 收口
            }
            next = avail + 1
        }
        p.seq.Set(avail) // 发布进度：生产者据此判断槽位可否复用
    }
}

type Disruptor struct {
    rb   *RingBuffer
    seqr Sequencer
    stop chan struct{}
}

// HandleWith：注册一个并行消费者，直接消费 ring buffer
func (d *Disruptor) HandleWith(wait WaitStrategy, on func(*Event, int64, bool)) *BatchProcessor {
    p := &BatchProcessor{barrier: &Barrier{seqr: d.seqr, wait: wait}, rb: d.rb, onEvent: on}
    d.seqr.AddGating(&p.seq)
    go p.Run(d.stop)
    return p
}

// After：注册链式消费者——prev 全部推进到位，才轮到它
func (d *Disruptor) After(prev []*BatchProcessor, wait WaitStrategy,
    on func(*Event, int64, bool)) *BatchProcessor {
    deps := make([]*Sequence, len(prev))
    for i, p := range prev {
        deps[i] = &p.seq
    }
    p := &BatchProcessor{barrier: &Barrier{seqr: d.seqr, deps: deps, wait: wait},
        rb: d.rb, onEvent: on}
    d.seqr.AddGating(&p.seq)
    go p.Run(d.stop)
    return p
}

// Publish：生产者侧两步——先写槽位，后发布（release 语义，07.md 的"发布 + 观察"）
func (d *Disruptor) Publish(ev *Event) {
    seq := d.seqr.Next()
    *d.rb.slot(seq) = *ev
    d.seqr.Publish(seq)
}

type State struct {
    // 订单簿、账户、持仓……全部在内存，2GB 量级装得下 2000 万槽位管线的输入
}

// Apply 必须是确定性的：同一事件序列 → 同一状态转移 + 同一输出
// 这是整个架构的地基：输出不落盘，靠重算再生
func (s *State) Apply(req Request) []OutputEvent {
    // 校验 → 改内存状态 → 产出输出事件（成交、确认、持仓变动……）
    return nil
}

// 记录格式：len(4B) | crc32(4B) | payload —— 自描述、可校验、崩溃可截断
type Journal struct {
    f *os.File
    w *bufio.Writer
}

var head [8]byte

func (j *Journal) Append(payload []byte) {
    binary.LittleEndian.PutUint32(head[0:4], uint32(len(payload)))
    binary.LittleEndian.PutUint32(head[4:8], crc32.ChecksumIEEE(payload))
    j.w.Write(head[:])
    j.w.Write(payload)
}

func (j *Journal) Sync() error { // 只在 endOfBatch 里调：一批一次 fsync
    if err := j.w.Flush(); err != nil {
        return err
    }
    return j.f.Sync()
}

// 写法是 08-5 的三步舞：写临时文件 → fsync → rename 原子转正
func (s *State) Snapshot(snapPath string, journalOffset int64) error {
    tmp := snapPath + ".tmp"
    f, err := os.Create(tmp)
    if err != nil {
        return err
    }
    defer f.Close()
    data := s.marshal()
    var head [24]byte
    binary.LittleEndian.PutUint32(head[0:4], snapMagic)
    binary.LittleEndian.PutUint32(head[4:8], snapVersion)
    binary.LittleEndian.PutUint64(head[8:16], uint64(journalOffset))
    binary.LittleEndian.PutUint64(head[16:24], uint64(len(data)))
    f.Write(head[:])
    f.Write(data)
    if err := f.Sync(); err != nil {
        return err
    }
    return os.Rename(tmp, snapPath) // 原子切换：崩溃前，旧快照永远完好
}

func Recover(snapPath, journalPath string) *State {
    st, off := loadSnapshot(snapPath) // 校验 magic/version/crc，坏了退上一代
    f, _ := os.Open(journalPath)
    f.Seek(off, io.SeekStart) // 只重放快照点之后

    r := bufio.NewReader(f)
    var head [8]byte
    for {
        if _, err := io.ReadFull(r, head[:]); err != nil {
            break // 尾部不足 8B = 崩溃现场，截断
        }
        n := binary.LittleEndian.Uint32(head[0:4])
        sum := binary.LittleEndian.Uint32(head[4:8])
        payload := make([]byte, n)
        if _, err := io.ReadFull(r, payload); err != nil ||
            crc32.ChecksumIEEE(payload) != sum {
            break // 半条或坏条：crc 失败处截断，之前的全部有效
        }
        st.Apply(unmarshal(payload)) // 重放：状态生效，输出丢弃
    }
    return st
}

bufSize := 1 << 21 // 209 万槽位演示；LMAX 输入 2000 万，输出每主题 400 万

d := NewDisruptor(bufSize, NewSingleProducer(int64(bufSize)))

// 并行组 1：journal / replicate / unmarshal，三个 I/O 密集消费者同时跑
journal := d.HandleWith(BusySpin{}, func(ev *Event, seq int64, eob bool) {
    j.Append(ev.Raw)
    if eob {
        j.Sync() // 一批一次 fsync：丢失窗口 ≤ 一批，I/O 成本摊薄到批粒度
    }
})
replica := d.HandleWith(BusySpin{}, func(ev *Event, seq int64, eob bool) {
    for _, s := range secondaries {
        s.SendWaitAck(ev.Raw) // 同步等 ACK：返回前，下游 BLP 被 gating 挡住
    }
})
unmarsh := d.HandleWith(BusySpin{}, func(ev *Event, seq int64, eob bool) {
    ev.Req = unmarshal(ev.Raw) // 写本槽 Req 字段：此字段唯一写者
})

// BLP：三个前置全部就位才消费。依赖图在这里是正确性约束——
// 回执发给客户之前，事件必须已落盘、已复制、已反序列化
d.After([]*BatchProcessor{journal, replica, unmarsh}, BusySpin{},
    func(ev *Event, seq int64, eob bool) {
        for _, out := range state.Apply(ev.Req) { // 单线程业务核心
            topics[out.Topic].Publish(out)        // 转投输出 Disruptor
        }
    })

// 输入侧生产者：网络读取，唯一 goroutine（单生产者，见 2.1 的理由）
func readLoop(conn net.Conn, d *Disruptor) {
    for {
        d.Publish(&Event{Raw: readMessage(conn)})
    }
}

type Topic struct{ d *Disruptor }

func (t *Topic) Publish(out OutputEvent) {
    seq := t.d.seqr.Next()
    t.d.rb.slot(seq).Out = out
    t.d.seqr.Publish(seq)
}

marsh := t.d.HandleWith(Yielding{}, marshalFn)         // 环 1：序列化
t.d.After([]*BatchProcessor{marsh}, Yielding{}, sendFn) // 环 2：网络发送