第22章 Go 常见设计模式

Go 没有类继承，也没有构造器重载，传统 OOP 设计模式（如 Factory、Abstract Factory）在 Go 里往往被简化甚至消解。但 Go 有自己的一套 idiomatic 模式，基于组合、函数值、channel 与 goroutine，解决配置、构建、并发编排等问题。本章覆盖工程中最常用的八种模式，每个都给出完整可运行示例。阅读前建议先掌握第6章函数与方法、第7章接口、第8章并发。

Option Pattern

1. 是什么

Option Pattern（配置结构模式）通过一个 options 结构体集中描述可选项，构造函数接收一个 *Options 指针并把零值视为默认。它是最朴素的“多可选参数“方案，也是 Functional Option 的基础。

2. 为什么这样设计 / 底层要点

Go 不支持函数重载，无法像 Java 那样为不同参数组合写多个构造器。直接用一长串参数会让调用方必须记住顺序且难以扩展。Options 结构体的优势：

零值即默认：调用方只需设置关心的字段。
新增字段不破坏调用方（向后兼容）。
类型安全，IDE 自动补全。

底层要点：结构体字段顺序影响内存对齐，把 bool/小类型放一起可省内存；对外暴露的 Options 应有清晰字段名与注释。

3. 工程实践与常见坑

完整示例：

package main

import (
	"fmt"
	"time"
)

// ServerOptions 描述 Server 的所有可选项。
// 零值即合理默认：Timeout=0 表示不超时，MaxConn=0 表示无限制。
type ServerOptions struct {
	Timeout  time.Duration
	MaxConn  int
	Verbose  bool
	OnClose  func()
}

// Server 是被构造的目标
type Server struct {
	opts ServerOptions
}

// NewServer 接收 options 指针；传 nil 用全默认
func NewServer(addr string, opts *ServerOptions) *Server {
	if opts == nil {
		opts = &ServerOptions{}
	}
	// 可选：补全默认值
	if opts.MaxConn == 0 {
		opts.MaxConn = 100
	}
	return &Server{opts: *opts}
}

func (s *Server) Run() {
	fmt.Printf("running with timeout=%v maxConn=%d verbose=%v\n",
		s.opts.Timeout, s.opts.MaxConn, s.opts.Verbose)
}

func main() {
	// 全默认
	NewServer(":8080", nil).Run()

	// 部分自定义
	NewServer(":8080", &ServerOptions{
		Timeout: 5 * time.Second,
		Verbose: true,
	}).Run()
}

常见坑：

指针 vs 值：传 *ServerOptions 让 nil 默认成为可能；若传值则调用方必须构造结构体。两种风格都常见，按库风格统一即可。
默认值补全：在构造函数里统一补全，避免业务代码各处 if x == 0。
0 值歧义：若 0 是合法业务值（如 MaxConn=0 表示“禁用连接“），需用 *int 或 sentinel 区分“未设置“与“设置为 0“。
可变性问题：上例把 opts 复制进 Server（opts: *opts），避免外部修改影响已构造对象；若 Options 含切片/map，浅拷贝仍共享底层数组，需注意。

Functional Option

1. 是什么

Functional Option 是 Go 社区最经典的配置模式（由 Dave Cheney、Rob Pike 等推广）：构造函数接收若干 Option 函数（type Option func(*config)），每个选项函数修改内部配置。调用方用 WithTimeout(t)、WithMaxConn(n) 这种风格串联。

2. 为什么这样设计 / 底层要点

相比 Option Pattern 的结构体，Functional Option 的额外优势：

调用更声明式：NewServer(addr, WithTimeout(t), WithVerbose())。
选项可封装在包内，默认值与校验逻辑集中在选项函数里。
第三方可自定义 Option 函数扩展（只要能拿到 *config）。
新增选项只是新增一个 WithXxx 函数，零破坏性。

底层要点：Option 是函数类型，闭包捕获参数。每个 WithXxx 返回一个闭包，在 NewServer 里依次应用到 config。这就是函数式编程中的“配置累加器“。

3. 工程实践与常见坑

完整示例：

package main

import (
	"fmt"
	"time"
)

// 内部配置，不对外暴露字段
type serverConfig struct {
	timeout time.Duration
	maxConn int
	verbose bool
	onClose func()
}

// Option 是配置函数
type Option func(*serverConfig)

// 选项函数
func WithTimeout(d time.Duration) Option {
	return func(c *serverConfig) { c.timeout = d }
}

func WithMaxConn(n int) Option {
	return func(c *serverConfig) {
		if n < 0 {
			panic("MaxConn must be non-negative")
		}
		c.maxConn = n
	}
}

func WithVerbose() Option {
	return func(c *serverConfig) { c.verbose = true }
}

func WithOnClose(f func()) Option {
	return func(c *serverConfig) { c.onClose = f }
}

// Server 对外类型
type Server struct {
	cfg serverConfig
}

// NewServer 接收可变长 Option
func NewServer(addr string, opts ...Option) *Server {
	cfg := serverConfig{
		timeout: 3 * time.Second, // 默认值
		maxConn: 100,
	}
	for _, opt := range opts {
		opt(&cfg)
	}
	return &Server{cfg: cfg}
}

func (s *Server) Run() {
	fmt.Printf("running: timeout=%v maxConn=%d verbose=%v\n",
		s.cfg.timeout, s.cfg.maxConn, s.cfg.verbose)
}

func main() {
	s := NewServer(":8080",
		WithTimeout(10*time.Second),
		WithMaxConn(500),
		WithVerbose(),
		WithOnClose(func() { fmt.Println("bye") }),
	)
	s.Run()
}

常见坑：

校验放选项函数里：如 WithMaxConn 校验负数，早失败早定位。也可在 NewServer 末尾统一校验。
默认值集中管理：在 NewServer 初始化 cfg 时给默认，避免“没传就零值“的歧义。
不要暴露 config 类型：把 serverConfig 改为小写未导出，调用方只能通过 WithXxx 操作，保证不变量。
选项顺序敏感时谨慎：若 WithXxx 之间有依赖（如 WithTLS 需要 WithCert 先设置），要么文档说明，要么在末尾统一解析。
性能：闭包有少量分配，构造期开销可忽略；热路径别用。

Functional Option 与 Option Pattern 可以混用：底层用结构体存配置，对外暴露 Functional Option API。这是 gRPC、Kubernetes client-go 的常见做法。

Builder

1. 是什么

Builder 模式把复杂对象的构造拆成多个步骤方法，每个方法返回 Builder 自身以支持链式调用，最后用 Build() 产出目标对象。适合“参数多、有构造中间状态、需要校验“的场景，如 SQL 构建、HTTP 请求构建、配置组装。

2. 为什么这样设计 / 底层要点

Go 没有命名参数与构造器链，Builder 用方法链弥补。要点：

Builder 持有可变中间状态，Build() 时一次性校验并构造不可变目标。
链式返回 *Builder 让 b.WithA().WithB().Build() 流畅。
目标对象可设计为不可变（字段小写未导出），保证构造后不被篡改。

与 Functional Option 的区别：Builder 更适合“分步骤、有顺序、有中间产物“的构造；Functional Option 更适合“一次性列举配置“。Builder 也常用于生成字符串/SQL 这类非对象结果。

3. 工程实践与常见坑

完整示例（HTTP 请求 Builder）：

package main

import (
	"errors"
	"fmt"
	"net/http"
	"strings"
)

// Request 是构建产物，字段未导出保证不可变
type Request struct {
	method  string
	url     string
	headers map[string]string
	body    string
}

func (r *Request) String() string {
	headers := make([]string, 0, len(r.headers))
	for k, v := range r.headers {
		headers = append(headers, k+": "+v)
	}
	return fmt.Sprintf("%s %s\n%s\nbody=%s", r.method, r.url,
		strings.Join(headers, "\n"), r.body)
}

// RequestBuilder 链式构建器
type RequestBuilder struct {
	method  string
	url     string
	headers map[string]string
	body    string
}

func NewRequestBuilder() *RequestBuilder {
	return &RequestBuilder{
		method:  "GET",
		headers: map[string]string{},
	}
}

func (b *RequestBuilder) Method(m string) *RequestBuilder {
	b.method = m
	return b
}

func (b *RequestBuilder) URL(u string) *RequestBuilder {
	b.url = u
	return b
}

func (b *RequestBuilder) Header(k, v string) *RequestBuilder {
	b.headers[k] = v
	return b
}

func (b *RequestBuilder) Body(s string) *RequestBuilder {
	b.body = s
	return b
}

// Build 校验并产出 Request
func (b *RequestBuilder) Build() (*Request, error) {
	if b.url == "" {
		return nil, errors.New("url is required")
	}
	if b.body != "" && b.method == http.MethodGet {
		return nil, errors.New("GET must not have body")
	}
	return &Request{
		method:  b.method,
		url:     b.url,
		headers: b.headers,
		body:    b.body,
	}, nil
}

func main() {
	r, err := NewRequestBuilder().
		Method("POST").
		URL("https://api.example.com/users").
		Header("Content-Type", "application/json").
		Body(`{"name":"go"}`).
		Build()
	if err != nil {
		fmt.Println("err:", err)
		return
	}
	fmt.Println(r)
}

常见坑：

Builder 可被复用导致状态污染：上例 Build() 后 Builder 仍可继续改字段再 Build，产出共享内部 map。若要禁止，可在 Build() 后置空，或文档约定一次性使用。
校验时机：在 Build() 统一校验，避免每个 WithXxx 都校验导致顺序耦合。
不可变目标：产物字段小写、不暴露 setter，确保“构造完即只读“。
链式断链：忘了 return b 会让链式调用编译失败，这是好事；用 receiver 指针 *Builder 而非值，否则修改不生效。

Pipeline

1. 是什么

Pipeline 模式把处理流程拆成多个 stage，每个 stage 是一个 goroutine，从前一个 channel 读入、处理后写到下一个 channel。数据像在管道里流动，天然并行：不同 stage 可在不同 CPU 上同时处理不同数据。

2. 为什么这样设计 / 底层要点

Pipeline 的核心是 channel 解耦生产与消费。底层要点：

每个 stage 是 func(in <-chan T) <-chan U：接收输入 channel，返回输出 channel，内部起 goroutine。
stage 之间通过 channel 传递，背压（backpressure）自然形成：下游慢，上游写不进去就阻塞。
错误传播：用单独的 errCh 或把结果包装成 Result{T, error} 在主 channel 流动。
取消传播：用 context.Context，任一 stage 失败 cancel ctx，所有 stage 退出。
资源释放：每个 stage 的 goroutine 必须在 channel 关闭或 ctx 取消时退出，否则泄漏。

经典三阶段 pipeline：generate → process → collect。

3. 工程实践与常见坑

完整示例（数字生成 → 平方 → 打印，带 ctx 取消）：

package main

import (
	"context"
	"fmt"
	"time"
)

// stage1: 生成数字
func generate(ctx context.Context, nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			select {
			case <-ctx.Done():
				return
			case out <- n:
			}
		}
	}()
	return out
}

// stage2: 平方
func square(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case <-ctx.Done():
				return
			case out <- n * n:
			}
		}
	}()
	return out
}

// stage3: 打印
func print(ctx context.Context, in <-chan int) {
	for n := range in {
		select {
		case <-ctx.Done():
			return
		default:
			fmt.Println(n)
		}
	}
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	out := square(ctx, generate(ctx, 1, 2, 3, 4, 5))
	print(ctx, out)
}

带 fan-out（多个 square 并行）的版本见 Fan-Out 小节。

常见坑：

goroutine 泄漏：stage 不监听 ctx 或不响应 in 关闭，下游不读时永远阻塞。务必 defer close(out) 且 select 包裹写。
背压变成阻塞：中间 channel 缓冲设太大失去背压作用；设 0（无缓冲）让上下游紧密同步，或设小缓冲（1-16）平滑波动。
错误处理：每个 stage 都可能失败。推荐用 Result 结构：

type Result struct {
	Value int
	Err   error
}

任一 stage 出错，把 Err 写入输出 channel，最终 stage 聚合并 cancel。

顺序性：pipeline 默认保持顺序（单 stage 单 goroutine）；若 fan-out 并行则乱序，需要带序号重排。

Pipeline 是 Go 并发的招牌模式，第8章并发中 “Share Memory by Communicating” 的具体落地。

Worker Pool

1. 是什么

Worker Pool 维护固定数量的 worker goroutine 处理任务队列。生产者把任务投递到 job channel，N 个 worker 并发消费，结果写入 result channel。它用于限制并发度、复用资源（如 DB 连接）、防止 goroutine 爆炸。

2. 为什么这样设计 / 底层要点

为什么不直接 go func() 每个任务？因为：

goroutine 虽轻量（2KB 起栈），但百万级会吃内存；且每个 goroutine 可能持有连接、文件句柄等重资源。
无限制并发会打垮下游（DB、第三方 API），需要限流。
Worker pool 让并发度 = N，可观测、可调优。

底层要点：

job channel 与 result channel 分离，解耦生产/消费。
用 sync.WaitGroup 等待所有 worker 完成。
优雅关闭：close(jobCh) 让 worker 处理完剩余任务后退出；ctx 取消则尽快退出。
worker 数通常 = runtime.NumCPU()（CPU 密集）或远大于 CPU 数（IO 密集，如 10-100）。

3. 工程实践与常见坑

完整示例：

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type Task struct {
	ID    int
	Input int
}

type Result struct {
	TaskID int
	Output int
	Err    error
}

// Pool 持有 worker 与 channel
type Pool struct {
	workers int
	jobs    chan Task
	results chan Result
	wg      sync.WaitGroup
}

func NewPool(workers, queueSize int) *Pool {
	return &Pool{
		workers: workers,
		jobs:    make(chan Task, queueSize),
		results: make(chan Result, queueSize),
	}
}

func (p *Pool) Start(ctx context.Context) {
	for i := 0; i < p.workers; i++ {
		p.wg.Add(1)
		go p.worker(ctx, i)
	}
	// 等所有 worker 退出后关闭 results
	go func() {
		p.wg.Wait()
		close(p.results)
	}()
}

func (p *Pool) worker(ctx context.Context, id int) {
	defer p.wg.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case task, ok := <-p.jobs:
			if !ok {
				return // jobs channel 关闭，优雅退出
			}
			res := Result{TaskID: task.ID}
			// 模拟处理
			time.Sleep(10 * time.Millisecond)
			res.Output = task.Input * task.Input
			p.results <- res
		}
	}
}

func (p *Pool) Submit(t Task) bool {
	select {
	case p.jobs <- t:
		return true
	default:
		return false // 队列满，拒绝
	}
}

func (p *Pool) Close() {
	close(p.jobs)
}

func (p *Pool) Results() <-chan Result {
	return p.results
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	pool := NewPool(4, 16)
	pool.Start(ctx)

	// 提交任务
	go func() {
		for i := 0; i < 20; i++ {
			pool.Submit(Task{ID: i, Input: i})
		}
		pool.Close() // 不再提交，触发 worker 优雅退出
	}()

	// 收集结果
	for r := range pool.Results() {
		fmt.Printf("task=%d out=%d\n", r.TaskID, r.Output)
	}
}

常见坑：

Submit 阻塞 vs 拒绝：p.jobs <- t 在队列满时阻塞生产者；select default 则拒绝。高可用系统倾向拒绝 + 上层重试，避免雪崩。
结果 channel 顺序：worker 并发消费，结果乱序；若需保序，给 Task 加序号在收集端重排。
panic 隔离：worker 内 panic 会让整个 pool 的 goroutine 死掉。worker 应 defer recover，把 panic 转成 Result.Err：

func (p *Pool) worker(ctx context.Context, id int) {
	defer p.wg.Done()
	defer func() {
		if r := recover(); r != nil {
			p.results <- Result{Err: fmt.Errorf("worker panic: %v", r)}
		}
	}()
	// ...
}

优雅关闭：close(jobs) 后 worker 处理完剩余任务再退出，wg.Wait() 保证不丢任务；ctx.Cancel 则是“尽快放弃“语义。
worker 数调优：参考第21章性能优化的 benchmark 与 pprof 定位最佳并发度。

Fan-In

1. 是什么

Fan-In（扇入）把多个输入 channel 的数据汇聚到一个输出 channel。多个生产者并行工作，结果汇总到单一消费者。它是“多对一“的合并模式。

2. 为什么这样设计 / 底层要点

Fan-In 的价值在于：消费者只需监听一个 channel，不必管理多个；多个生产者可并行提速。底层实现两种方式：

每个输入起一个 goroutine 转发（简单，goroutine 数 = 输入数）。
reflect.Select 动态监听多 channel（避免多 goroutine，但 reflect 有开销，且 case 数有上限）。

goroutine 转发法最常用：对每个 in channel 起一个 goroutine，把数据搬到 out channel，所有 goroutine 退出后 close(out)。用 sync.WaitGroup 同步。

3. 工程实践与常见坑

完整示例（多源搜索合并）：

package main

import (
	"context"
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// 模拟一个数据源
func source(ctx context.Context, name string, n int) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for i := 0; i < n; i++ {
			select {
			case <-ctx.Done():
				return
			case out <- fmt.Sprintf("%s-%d", name, i):
				time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond)
			}
		}
	}()
	return out
}

// FanIn 合并多个 channel
func FanIn(ctx context.Context, channels ...<-chan string) <-chan string {
	var wg sync.WaitGroup
	out := make(chan string)

	transfer := func(c <-chan string) {
		defer wg.Done()
		for v := range c {
			select {
			case <-ctx.Done():
				return
			case out <- v:
			}
		}
	}

	wg.Add(len(channels))
	for _, c := range channels {
		go transfer(c)
	}

	// 所有 transfer 结束后关闭 out
	go func() {
		wg.Wait()
		close(out)
	}()
	return out
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	// 三个数据源并行
	merged := FanIn(ctx,
		source(ctx, "db", 5),
		source(ctx, "cache", 5),
		source(ctx, "api", 5),
	)

	for v := range merged {
		fmt.Println(v)
	}
}

常见坑：

关闭时机：必须等所有 transfer goroutine 退出后再 close(out)，否则消费者读到关闭 channel 时还有数据未搬完。用 wg.Wait() 保证。
ctx 取消语义：取消后 transfer 应尽快退出，但 out 可能仍被消费者读；close(out) 仍由 wg 触发，安全。
死锁：若消费者不读 out，transfer 写 out 阻塞，wg 永不结束，out 永不关闭。给 out 设缓冲或确保消费者及时消费。
顺序丢失：FanIn 后数据交错，不保持各源内部顺序之外的任何顺序；需要顺序时改用 Fan-In + 序号重排或串行化。
reflect.Select 的取舍：输入 channel 数固定且少（<10）用 goroutine 法；输入动态增减才考虑 reflect.Select 或 sync + map。

Fan-Out

1. 是什么

Fan-Out（扇出）把一个输入 channel 分发给多个并行的 worker 处理，是“一对多“的扩散模式。常与 Fan-In 组合成 Fan-Out/Fan-In：一个任务分发给 N 个 worker 并行处理，结果再汇聚。

2. 为什么这样设计 / 底层要点

Fan-Out 的目的是并行化处理。底层关键：

多个 worker 共享同一个输入 channel（竞争消费），Go runtime 自动负载均衡——谁空闲谁拿。
每条消息只被一个 worker 处理（work-stealing 语义），不同于 Pub/Sub 的广播。
并行度 = worker 数，可控。
结果若需汇聚，用 Fan-In 把各 worker 的输出 channel 合并。

与 Worker Pool 的关系：Fan-Out 本质就是“共享 job channel 的 worker pool“，视角不同。Worker Pool 强调“池化管理“，Fan-Out 强调“任务扩散“。

3. 工程实践与常见坑

完整示例（Fan-Out + Fan-In：图片处理流水线）：

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

type Image struct {
	ID   int
	Size int
}

type ProcessedImage struct {
	ID     int
	Bytes  int
	Worker int
}

// 模拟处理：耗时与 size 正比
func process(img Image, workerID int) ProcessedImage {
	time.Sleep(time.Duration(img.Size) * time.Millisecond)
	return ProcessedImage{ID: img.ID, Bytes: img.Size * 2, Worker: workerID}
}

// FanOut: 启动 N 个 worker 竞争消费 in，输出到各自的 out，再 FanIn
func FanOutFanIn(ctx context.Context, in <-chan Image, workers int) <-chan ProcessedImage {
	outs := make([]chan ProcessedImage, workers)
	for i := range outs {
		outs[i] = make(chan ProcessedImage)
	}

	// Fan-Out: N 个 worker 共享 in
	var wg sync.WaitGroup
	wg.Add(workers)
	for w := 0; w < workers; w++ {
		go func(id int) {
			defer wg.Done()
			defer close(outs[id])
			for img := range in {
				select {
				case <-ctx.Done():
					return
				case outs[id] <- process(img, id):
				}
			}
		}(w)
	}

	// Fan-In: 合并所有 out
	merged := make(chan ProcessedImage)
	go func() {
		var fin sync.WaitGroup
		fin.Add(len(outs))
		for _, o := range outs {
			go func(c <-chan ProcessedImage) {
				defer fin.Done()
				for v := range c {
					merged <- v
				}
			}(o)
		}
		fin.Wait()
		close(merged)
	}()

	return merged
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	// 生产图片
	in := make(chan Image)
	go func() {
		defer close(in)
		for i := 0; i < 10; i++ {
			in <- Image{ID: i, Size: 20}
		}
	}()

	// 4 worker 并行处理
	for r := range FanOutFanIn(ctx, in, 4) {
		fmt.Printf("img=%d bytes=%d by worker=%d\n", r.ID, r.Bytes, r.Worker)
	}
}

常见坑：

worker 不均：负载均衡靠 runtime，但如果某些任务特别慢，会拖慢整体。可考虑按任务大小分桶或用 work-stealing 队列。
结果顺序：并发处理导致结果乱序，需要保序时加序号重排。
goroutine 数控制：worker 数 = 并发度，别开太多；FanIn 的合并 goroutine 数 = worker 数，可控。
关闭传播：in 关闭后所有 worker 自然退出（range 结束），各 out 关闭，FanIn 检测到所有 out 关闭后关闭 merged，链条完整。
错误处理：把 ProcessedImage 改成 Result{T, error}，worker 出错时写 Result.Err，FanIn 透传，最终聚合决定 cancel 或跳过。
背压：in channel 无缓冲或小缓冲让生产者跟随消费速率；merged channel 同理。

Pub/Sub

1. 是什么

Pub/Sub（发布/订阅）模式：发布者把消息发到一个 topic，多个订阅者各收到一份副本（广播）。发布者不关心谁订阅，订阅者不关心谁发布，二者通过 broker 解耦。与 Fan-Out 的区别：Fan-Out 是“一份消息给一个 worker“，Pub/Sub 是“一份消息给所有订阅者“。

2. 为什么这样设计 / 底层要点

Go 内实现 Pub/Sub 的关键：每个订阅者有自己的 channel，broker 维护订阅者列表，收到消息时遍历转发。底层要点：

订阅者 channel 缓冲策略：无缓冲会阻塞发布者（慢订阅者拖垮整体）；有缓冲满则丢或阻塞，需权衡。
慢消费者策略：阻塞（保数据但风险阻塞发布者）、丢弃旧消息（default 不写）、断开订阅者。常见选“丢弃旧消息 + 监控“。
取消订阅：从订阅者列表移除并关闭其 channel，让消费者 range 退出。
并发安全：订阅/退订/发布并发，broker 用 sync.RWMutex 保护订阅者列表。

3. 工程实践与常见坑

完整示例（线程安全的内存 Pub/Sub broker）：

package main

import (
	"fmt"
	"sync"
	"time"
)

type Event struct {
	Topic string
	Data  interface{}
}

type subscription struct {
	ch chan Event
}

// Broker 是 Pub/Sub 核心
type Broker struct {
	mu     sync.RWMutex
	subs   map[string][]*subscription // topic -> subscribers
	closed bool
}

func NewBroker() *Broker {
	return &Broker{subs: make(map[string][]*subscription)}
}

// Subscribe 订阅 topic，返回只读 channel
func (b *Broker) Subscribe(topic string, buf int) <-chan Event {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		ch := make(chan Event)
		close(ch)
		return ch
	}
	s := &subscription{ch: make(chan Event, buf)}
	b.subs[topic] = append(b.subs[topic], s)
	return s.ch
}

// Publish 发布消息。慢消费者：丢弃最旧消息（非阻塞）
func (b *Broker) Publish(topic string, data interface{}) {
	b.mu.RLock()
	defer b.mu.RUnlock()
	if b.closed {
		return
	}
	for _, s := range b.subs[topic] {
		select {
		case s.ch <- Event{Topic: topic, Data: data}:
		default:
			// 缓冲满，丢弃（生产环境应打指标）
			fmt.Printf("WARN: drop msg on topic=%s\n", topic)
		}
	}
}

// Close 关闭所有订阅，退出所有消费者
func (b *Broker) Close() {
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.closed {
		return
	}
	b.closed = true
	for _, subs := range b.subs {
		for _, s := range subs {
			close(s.ch)
		}
	}
	b.subs = make(map[string][]*subscription)
}

func main() {
	broker := NewBroker()

	// 两个订阅者
	sub1 := broker.Subscribe("news", 8)
	sub2 := broker.Subscribe("news", 8)

	var wg sync.WaitGroup
	wg.Add(2)
	consume := func(name string, ch <-chan Event) {
		defer wg.Done()
		for e := range ch {
			fmt.Printf("%s got: %v\n", name, e.Data)
		}
		fmt.Printf("%s done\n", name)
	}
	go consume("sub1", sub1)
	go consume("sub2", sub2)

	// 发布
	go func() {
		for i := 0; i < 5; i++ {
			broker.Publish("news", fmt.Sprintf("headline-%d", i))
			time.Sleep(50 * time.Millisecond)
		}
		broker.Close() // 触发消费者退出
	}()

	wg.Wait()
}

常见坑：

慢消费者策略选择：
- 阻塞（去掉 default）：保证不丢但慢订阅者拖垮发布者与其它订阅者。
- 丢弃（default）：保发布者吞吐但丢数据，适合指标/日志这类容忍丢失的场景。
- 断开订阅者：发布者检测阻塞后从列表移除该订阅者，复杂但公平。生产系统常配指标观察丢弃率与订阅延迟，按场景选。
订阅者列表并发修改：Publish 用 RLock（多读并发），Subscribe/Unsubscribe 用 Lock。注意 Publish 持有 RLock 时不能调用会改列表的方法，否则死锁。
关闭后发布：Close 设 closed 标志，Publish 检查后直接返回，避免写已关闭 channel panic。
重复订阅：同一消费者 Subscribe 两次会收到两份，业务层避免。
跨进程 Pub/Sub：本节是进程内 broker。分布式场景用 NATS、Kafka、Redis Pub/Sub，Go 内用接口抽象便于切换。
背压与缓冲：缓冲大小需根据消息速率与消费者处理速度压测确定；监控 channel 长度是关键指标。
泛型版本：Go 1.18+ 可用泛型 Broker[T] 让 Event 携带类型安全数据，避免 interface{} 类型断言。

本章小结

Go 的设计模式强调组合优于继承、显式优于隐式。本章八种模式可分两组：

配置与构造组：

Option Pattern：结构体集中可选项，零值即默认，简单直接。
Functional Option：WithXxx 函数列表，声明式、可扩展、可校验，社区主流。
Builder：分步链式构造，适合多步骤、需校验的复杂对象/字符串构建。

并发编排组：

Pipeline：stage 串联，channel 解耦，背压自然，Go 并发招牌。
Worker Pool：固定 worker 处理任务队列，限流复用，防 goroutine 爆炸。
Fan-In：多输入合并为单输出，消费者统一处理。
Fan-Out：单输入分发给多 worker 并行，常与 Fan-In 组合。
Pub/Sub：一份数据广播给所有订阅者，发布/订阅解耦。

工程实践上的共性原则：

channel 与 goroutine 必须成对管理生命周期：谁创建谁关闭，用 ctx 传播取消，defer close 防泄漏。参考第8章并发。
错误是值：并发模式中错误走 channel 与数据一起流动，最终聚合处理，见第20章错误处理。
可观测：worker 数、channel 长度、丢弃率都是关键指标，配合第21章性能优化的 pprof/trace 调优。
先正确再性能：模式选型以代码清晰为先，热点出现后再用 benchmark 与 pprof 驱动优化，避免过早抽象。

掌握这些模式，你就具备了用 Go 写出可维护、可扩展、高性能并发系统的工程能力。接下来在第28章 Kubernetes 工程实践中，你会看到 client-go、controller-runtime 大量使用 Functional Option 与 Worker Pool，这些模式是云原生 Go 代码的通用语汇。

Keyboard shortcuts

Go 底层原理与工程实践