第25章 Go 常见设计模式

Go 没有类继承，也没有构造器重载，传统 OOP 设计模式在 Go 里往往被简化。Go 更常用组合、函数值、channel 与 goroutine 解决配置、构建和并发编排。阅读前建议先掌握第10章函数、第7章 Interface和第18章内存模型。

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 让交接同步发生；有界缓冲可吸收短时波动，但容量过大会推迟背压并增加在途内存。容量应由允许的在途任务数、单项大小、服务时间和延迟预算推导，不使用固定的 1-16 经验阈值。
错误处理：每个 stage 都可能失败。推荐用 Result 结构：

type Result struct {
	Value int
	Err   error
}

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

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

Pipeline 是 Go 并发的代表模式，其同步保证见第18章 Go 内存模型与数据竞争。

Worker Pool

1. 是什么

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

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

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

goroutine 相对线程轻量；自 Go 1.19 起，新 goroutine 的初始栈大小按历史 goroutine 的平均栈用量自适应调整（runtime 变量 startingStackSize，每轮 GC 更新一次），2 KiB（stackMin）只是下限而非固定起步值。此外调度状态、增长后的栈、闭包和所持连接等都会增加成本，初始值也不是语言契约。
无限制并发会打垮下游（DB、第三方 API），需要限流。
Worker pool 让并发度 = N，可观测、可调优。

底层要点：

job channel 与 result channel 分离，解耦生产/消费。
用 sync.WaitGroup 等待所有 worker 完成。
优雅关闭：close(jobCh) 让 worker 处理完剩余任务后退出；ctx 取消则尽快退出。
worker 数应由 CPU 配额、任务服务时间、下游并发上限和目标排队延迟共同决定。CPU 密集任务可从当前 GOMAXPROCS 附近开始实验；IO 密集也不能套用 10-100 或 CPU 倍数公式。

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)

	// 提交任务：Submit 返回 false 表示队列满被拒绝，必须处理，否则任务静默丢失
	go func() {
		for i := 0; i < 20; i++ {
			t := Task{ID: i, Input: i}
			for !pool.Submit(t) {
				// 生产环境可重试、降级或打指标，这里简单退避后重试
				fmt.Printf("task=%d rejected, backoff\n", t.ID)
				time.Sleep(20 * time.Millisecond)
			}
		}
		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 若不恢复，会顺着 goroutine 崩掉整个进程。注意 recover 不能只放在 worker 顶层——那样虽然进程不崩，但该 worker goroutine 会随 defer 结束而退出，池容量永久减一。正确做法是把“处理单个任务“包进一个闭包，在闭包里 recover，worker 循环继续存活；同时 recover 分支里的结果发送要用 select 监听 ctx，避免消费者已离开时永久阻塞：

func (p *Pool) worker(ctx context.Context, id int) {
	defer p.wg.Done()
	handle := func(task Task) (res Result) {
		defer func() {
			if r := recover(); r != nil {
				res = Result{TaskID: task.ID, Err: fmt.Errorf("worker panic: %v", r)}
			}
		}()
		res = Result{TaskID: task.ID}
		res.Output = task.Input * task.Input // 可能 panic 的业务逻辑
		return res
	}
	for {
		select {
		case <-ctx.Done():
			return
		case task, ok := <-p.jobs:
			if !ok {
				return
			}
			select {
			case p.results <- handle(task):
			case <-ctx.Done():
				return
			}
		}
	}
}

优雅关闭：close(jobs) 后 worker 处理完剩余任务再退出，wg.Wait() 保证不丢任务；ctx.Cancel 则是“尽快放弃“语义。
worker 数调优：参考第24章性能优化的 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 的取舍：固定输入可直接写 select，也可像示例一样为每个输入启动转发 goroutine；动态输入才需要评估 reflect.Select、集中式调度或其他注册机制。不存在通用的“少于 10 个”阈值，应按输入数量、生命周期、流量和分配实测。

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
	for w := 0; w < workers; w++ {
		go func(id int) {
			defer close(outs[id])
			for img := range in {
				// 注意：select 的 send case 右侧表达式在进入 select 时即求值，
				// 即使 ctx 已取消，process 也会先执行一次。先求值再 select 使语义显式。
				p := process(img, id)
				select {
				case <-ctx.Done():
					return
				case outs[id] <- p:
				}
			}
		}(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 {
					select {
					case <-ctx.Done():
						return
					case merged <- v:
					}
				}
			}(o)
		}
		fin.Wait()
		close(merged)
	}()

	return merged
}

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

	// 生产图片：必须监听 ctx，否则取消后 worker 都退出了，
	// 生产者在无缓冲 channel 上永久阻塞，goroutine 泄漏
	in := make(chan Image)
	go func() {
		defer close(in)
		for i := 0; i < 10; i++ {
			select {
			case <-ctx.Done():
				return
			case 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  any
}

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 any) {
	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 携带类型安全数据，避免 any 类型断言。

单例与惰性初始化：sync.OnceFunc / OnceValue（Go 1.21+）

1. 是什么

单例与惰性初始化在 Go 里传统写法是 sync.Once + 包级变量。Go 1.21 新增了 sync.OnceFunc、sync.OnceValue、sync.OnceValues 三个包装函数，把“只执行一次“直接封装成可调用的函数值，省去手写 once.Do 与共享变量。

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

OnceValue(f func() T) func() T 返回的函数并发安全：多个 goroutine 同时首次调用时只有一个执行 f，其余等待并共享结果。
若 f panic，之后每次调用都以同一个值重新 panic——失败不会被静默吞掉，也不会重试。
相比手写 sync.Once，闭包把“初始化逻辑 + 结果存储 + 同步“三者绑在一个值里，不再暴露可被绕过的包级变量，语义更内聚。

3. 工程实践与常见坑

package main

import (
	"fmt"
	"sync"
)

type Config struct {
	DSN string
}

// 并发安全的惰性单例：首次调用才真正加载
var loadConfig = sync.OnceValue(func() *Config {
	fmt.Println("loading config (only once)")
	return &Config{DSN: "mysql://localhost:3306/app"}
})

func main() {
	var wg sync.WaitGroup
	for i := 0; i < 4; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = loadConfig() // 只有一个 goroutine 执行初始化
		}()
	}
	wg.Wait()
	fmt.Println("dsn:", loadConfig().DSN)
}

常见坑：

panic 即“永久失败“：f panic 后没有重试机会。初始化可能失败且需要重试时，应改用返回 error 的普通函数加显式重试逻辑，而不是 OnceValue。
返回共享指针：OnceValue 返回同一个 *Config，所有调用方共享；若对象可变，仍需自行保证并发安全或返回副本。
别滥用为全局状态入口：惰性单例适合真正只读且昂贵的初始化（配置、正则编译、连接池）；业务依赖建议仍走显式注入（见第 33 章）。

errgroup：带错误传播的结构化并发

1. 是什么

golang.org/x/sync/errgroup 是官方扩展库中最常用的并发原语：Group 管理一组子任务 goroutine，Wait 等待全部结束并返回第一个非 nil 错误；配合 errgroup.WithContext，任一任务失败会取消共享 ctx，通知其余任务尽快退出。它把“WaitGroup + 错误收集 + 取消传播“三件事合成一个抽象。

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

g.Go(f) 内部即 wg.Add(1) + goroutine + 首个错误的 sync.Once 记录；Wait 即 wg.Wait() 后 cancel ctx 并返回该错误。
g.SetLimit(n)（x/sync v0.7+）限制同时运行的 goroutine 数，超限的 Go 调用阻塞，天然充当轻量 worker pool。
errgroup 不会替你捕获 f 的 panic（x/sync 曾在 v0.11 尝试把 panic 转发到 Wait，后因掩盖崩溃现场、可能死锁等问题在新版本中移除，源码注释引用了 #53757 等 issue）。需要 panic 隔离时按 Worker Pool 一节的方式在 f 内自行 recover。

3. 工程实践与常见坑

典型用法——并发抓取多个 URL，任一失败即取消其余：

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"golang.org/x/sync/errgroup"
)

func fetch(ctx context.Context, url string) (string, error) {
	select {
	case <-time.After(50 * time.Millisecond):
		if url == "https://bad.example.com" {
			return "", errors.New("fetch failed: " + url)
		}
		return "ok:" + url, nil
	case <-ctx.Done():
		return "", ctx.Err()
	}
}

func main() {
	urls := []string{
		"https://a.example.com",
		"https://bad.example.com",
		"https://b.example.com",
	}
	g, ctx := errgroup.WithContext(context.Background())
	g.SetLimit(2) // 限制并发度，兼作轻量 worker pool
	results := make([]string, len(urls))
	for i, u := range urls {
		g.Go(func() error {
			r, err := fetch(ctx, u)
			results[i] = r // 各 goroutine 写不同下标，无数据竞争
			return err
		})
	}
	if err := g.Wait(); err != nil {
		fmt.Println("group err:", err)
	}
	fmt.Println("results:", results)
}

常见坑：

只保留第一个错误：后续错误被丢弃。需要收集全部错误时，让每个任务把错误写入自己的下标，Wait 后用 errors.Join 聚合。
子任务必须尊重 ctx：errgroup 只负责 cancel，不会强杀 goroutine；f 里不监听 ctx 依旧会拖到自然结束。
循环变量：Go 1.22 起 for 循环变量按迭代绑定，闭包直接捕获 i、u 是安全的；1.21 及更早版本需 i, u := i, u。
相关原语：需要“n 个并发配额、任务权重不同“的限流时用 golang.org/x/sync/semaphore 的带权信号量（Acquire(ctx, weight) 可被 ctx 取消）；需要在 ctx 取消时触发一次性清理回调，可用 context.AfterFunc(ctx, f)（Go 1.21+），它返回的 stop 函数可解除注册。

本章小结

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

配置与构造组：

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

并发编排组：

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

此外，标准库与官方扩展库已把若干模式“产品化“：sync.OnceFunc/OnceValue（Go 1.21+）覆盖单例与惰性初始化，errgroup（配合 SetLimit）覆盖大部分“并发子任务 + 错误传播“场景，semaphore.Weighted 提供带权限流，context.AfterFunc（Go 1.21+）处理取消回调。写新代码时优先考虑这些现成原语，再考虑手写 channel 编排。

工程实践上的共性原则：

channel 与 goroutine 必须成对管理生命周期：由能证明不会再发送的协调者关闭数据 channel，用 ctx 传播取消，并等待派生 goroutine 退出。创建者、发送者和关闭者可能是同一方，也可能不是。参考第18章 Go 内存模型与数据竞争。
错误是值：并发模式中错误走 channel 与数据一起流动，最终聚合处理，见第23章错误处理。
可观测：worker 数、channel 长度、丢弃率都是关键指标，配合第24章性能优化的 pprof/trace 调优。
先正确再性能：模式选型以代码清晰为先，热点出现后再用 benchmark 与 pprof 驱动优化，避免过早抽象。

掌握这些模式后，可以继续阅读第29章 client-go与第30章 Controller，观察 Functional Option、队列和 Worker Pool 在云原生项目中的使用。

Keyboard shortcuts

Go 底层原理与工程实践