Reactor 与 Proactor 模式

概念定义

“一个线程管理成千上万 I/O”有两种答案。Reactor 等就绪并派发动作；Proactor 提交带缓冲的异步操作，等内核报告“已完成”。POSIX 世界以 Reactor 为主，Windows IOCP 原生 Proactor。

工作机制

Reactor：按描述符注册处理器，就绪即派发，处理器自己做非阻塞读写并处理 EAGAIN 与部分传输。Proactor：提交操作加缓冲，内核 DMA 填排，完成队列交付结果。io_uring 用提交环与完成环让 Proactor 风格在 Linux 上变得自然。

使用场景

设计网络运行时与理解缓冲生命周期时它才真正要紧：reactor 处理器在调用期间拥有缓冲；proactor 缓冲从提交到完成被钉住——这是无数微妙生命周期 bug 的温床。

常见误区

两者不是同义词：reactor 通知就绪你来搬，proactor 通知完成内核已搬。reactor 循环串行化——一个慢处理器拖延所有就绪描述符，计算必须外溢；proactor 缓冲在完成前复用会腐蚀在途数据，取消语义各平台不同且常常照常完成。错误上报是分裂的：就绪错误在派发处、完成错误在结果里，而生态文档常把它们混为一谈。

代码示例

export type Release = () => void;

export class ReadWriteLock {
  private readers = 0;
  private writer = false;
  private waitingReaders: Array<() => void> = [];
  private waitingWriters: Array<() => void> = [];

  async acquireRead(): Promise<Release> {
    if (!this.writer && this.waitingWriters.length === 0) {
      this.readers += 1;
      return this.createReadRelease();
    }
    await new Promise<void>((resolve) => this.waitingReaders.push(resolve));
    return this.createReadRelease();
  }

  async acquireWrite(): Promise<Release> {
    if (!this.writer && this.readers === 0 && this.waitingWriters.length === 0) {
      this.writer = true;
      return this.createWriteRelease();
    }
    await new Promise<void>((resolve) => this.waitingWriters.push(resolve));
    return this.createWriteRelease();
  }

  private createReadRelease(): Release {
    let released = false;
    return () => {
      if (released) return;
      released = true;
      this.readers -= 1;
      if (this.readers === 0 && this.waitingWriters.length > 0) {
        this.writer = true;
        this.waitingWriters.shift()!();
      }
    };
  }

  private createWriteRelease(): Release {
    let released = false;
    return () => {
      if (released) return;
      released = true;
      this.writer = false;
      if (this.waitingWriters.length > 0) {
        this.writer = true;
        this.waitingWriters.shift()!();
        return;
      }
      const readers = this.waitingReaders.splice(0);
      this.readers += readers.length;
      for (const resolve of readers) resolve();
    };
  }

  get state(): {
    readers: number;
    writer: boolean;
    waitingReaders: number;
    waitingWriters: number;
  } {
    return {
      readers: this.readers,
      writer: this.writer,
      waitingReaders: this.waitingReaders.length,
      waitingWriters: this.waitingWriters.length,
    };
  }
}

package main

import (
	"fmt"
	"sync"
)

type Broker struct {
	mu          sync.RWMutex
	subscribers map[int]chan string
	nextID      int
}

func NewBroker() *Broker {
	return &Broker{subscribers: make(map[int]chan string)}
}

func (b *Broker) Subscribe(buffer int) (int, <-chan string) {
	b.mu.Lock()
	defer b.mu.Unlock()
	id := b.nextID
	b.nextID++
	channel := make(chan string, buffer)
	b.subscribers[id] = channel
	return id, channel
}

func (b *Broker) Unsubscribe(id int) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if channel, ok := b.subscribers[id]; ok {
		delete(b.subscribers, id)
		close(channel)
	}
}

func (b *Broker) Publish(event string) int {
	b.mu.RLock()
	defer b.mu.RUnlock()
	delivered := 0
	for _, channel := range b.subscribers {
		select {
		case channel <- event:
			delivered++
		default:
		}
	}
	return delivered
}

func main() {
	broker := NewBroker()
	fastID, fast := broker.Subscribe(4)
	defer broker.Unsubscribe(fastID)
	slowID, slow := broker.Subscribe(1)
	defer broker.Unsubscribe(slowID)

	delivered := []int{broker.Publish("a"), broker.Publish("b"), broker.Publish("c")}
	fmt.Println("delivered:", delivered)
	if delivered[0] != 2 || delivered[1] != 1 || delivered[2] != 1 {
		panic("unexpected delivery counts")
	}

	fastSeen := []string{<-fast, <-fast, <-fast}
	slowSeen := []string{<-slow}
	fmt.Println(fastSeen, slowSeen)
	if slowSeen[0] != "a" {
		panic("slow subscriber should keep only the first event")
	}

	broker.Unsubscribe(slowID)
	if broker.Publish("d") != 1 {
		panic("expected one subscriber after unsubscribe")
	}
	broker.Unsubscribe(fastID)
	if broker.Publish("e") != 0 {
		panic("expected no subscribers")
	}
}