modern-go/concurrent 源码解析：从 concurrent.Map 到可取消的 Executor 并发模型

• 机器学习
• 深度学习
• 数据可视化
• 可观测性

modern-go/concurrent是一个轻量级的 Go 并发工具库，以两个核心 API 著称：concurrent.Map（sync.Map 的向后移植版本，解决 Go 1.9 以下版本的可移植性问题）与concurrent.Executor（为 goroutine 提供显式所有权、统一取消与 panic 兜底机制）。本指南将以本仓库中 vendored 的源码（core/vendor/github.com/modern-go/concurrent）为蓝本，逐一拆解两个 API 的使用方式、底层实现与设计动机，帮助你掌握"如何让 goroutine 变得可控、可取消、可追踪"的工程实践。

一、库概览：两个 API 解决什么问题

该库定位极小，只提供两个核心抽象：

此外，库内还暴露了两个可配置的全局日志器：ErrorLogger（默认输出到 stderr）与InfoLogger（默认丢弃），用于 executor 运行期间的错误与信息上报（见 log.go）。

二、concurrent.Map：sync.Map 的向后移植

1. 为什么需要它

sync.Map在 Go 1.9 中才被引入。若项目需要兼容 Go 1.9 以下的编译器，直接使用sync.Map会导致编译失败。concurrent.Map通过构建标签（build tags）在编译期自动选择正确的实现，使代码一次编写、处处可编译。

2. 双实现机制（构建标签分流）

同一目录下存在两个互为镜像的实现文件，通过// +build标签隔离：

• Go 1.9 及以上（go_above_19.go）：Map直接内嵌标准库的sync.Map，零成本复用官方实现：

// +build go1.9 type Map struct { sync.Map } func NewMap() *Map { return &Map{} }

• Go 1.9 以下（go_below_19.go）：以sync.RWMutex+ 普通 map 自行实现线程安全读写，读操作加读锁、写操作加写锁，语义与sync.Map对齐：

// +build !go1.9 type Map struct { lock sync.RWMutex data map[interface{}]interface{} } func NewMap() *Map { return &Map{ data: make(map[interface{}]interface{}, 32), } } func (m *Map) Load(key interface{}) (elem interface{}, found bool) { m.lock.RLock() elem, found = m.data[key] m.lock.RUnlock() return } func (m *Map) Store(key interface{}, elem interface{}) { m.lock.Lock() m.data[key] = elem m.lock.Unlock() }

注意两个细节：一是低版本实现为 map 预分配了 32 个桶位的初始容量，降低扩容频率；二是Store在锁内直接覆盖旧值，Load返回(elem, found)双返回值，与sync.Map的调用约定完全一致。

3. 使用方法（继承自 README）

无论最终编译到哪个 Go 版本，业务代码的写法完全一致：

m := concurrent.NewMap() m.Store("hello", "world") elem, found := m.Load("hello") // elem 为 "world" // found 为 true

从源码结构看，Map通过类型内嵌sync.Map（Go ≥ 1.9）继承了LoadOrStore、Delete、Range等其余方法；在低版本分支中则仅实现了Load/Store两个最小方法，因此跨版本代码建议只依赖这两个方法，以保证可移植性边界清晰。

三、concurrent.Executor：显式所有权与可取消的 goroutine

1. 设计动机：go关键字的缺陷

裸go func()启动的 goroutine 存在三个工程痛点：无法从外部统一取消；panic 一旦逃逸会直接崩溃整个进程；goroutine 与启动者之间没有归属关系，难以跟踪生命周期。Executor正是针对这三点设计的替代品。

接口定义非常克制（见 executor.go）：

type Executor interface { // Go starts a new goroutine controlled by the context Go(handler func(ctx context.Context)) }

接口只暴露Go方法；停止、等待等管理能力属于具体实现类型（如UnboundedExecutor），文档注释也明确提示：启动并拥有 executor 的一方应当使用具体类型而非接口。

2. UnboundedExecutor：不限制 goroutine 数量的实现

UnboundedExecutor（unbounded_executor.go）是对活跃 goroutine 数量不加上限的实现，其内部状态包括：

• 一个context.Context+cancel函数：作为统一取消信号源；
• 一个map[string]int+ 互斥锁：记录每个 goroutine 的启动位置（文件:行号）与存活计数，用于"等待全部退出"与问题排查；
• 可覆盖的HandlePanic回调：默认行为是打日志而非崩溃。

它还提供了一个进程级单例GlobalUnboundedExecutor，生命周期与程序本身对齐，适合承载希望在main退出前统一回收的后台任务。

3. 官方 README 示例（完整继承）

executor := concurrent.NewUnboundedExecutor() executor.Go(func(ctx context.Context) { everyMillisecond := time.NewTicker(time.Millisecond) for { select { case <-ctx.Done(): fmt.Println("goroutine exited") return case <-everyMillisecond.C: // do something } } }) time.Sleep(time.Second) executor.StopAndWaitForever() fmt.Println("executor stopped")

把 goroutine 挂到 executor 实例上之后，我们可以获得两种能力（README 原意）：

• 统一取消：通过Stop/StopAndWait/StopAndWaitForever停止 executor，从而取消其名下所有 goroutine；
• panic 兜底：goroutine 内发生的 panic 由回调统一处理，默认行为是记录日志，不再导致整个应用崩溃。

四、源码级机制拆解

1. Go：启动、追踪、兜底三合一

Go方法（unbounded_executor.go）执行三个关键步骤：

1. 记录启动位置：通过reflect.ValueOf(handler).Pointer()+runtime.FuncForPC解析出函数名、文件与行号，并在activeGoroutines计数中 +1；
2. 启动带 recover 的 goroutine：真正执行的handler(executor.ctx)外层包裹defer recover()，任何 panic 都会被捕获；
3. panic 分流处理：恢复后若recovered != nil，优先调用实例级executor.HandlePanic（若被显式覆盖），否则回落到包级全局HandlePanic；随后将存活计数 -1。

包级默认的HandlePanic实现（unbounded_executor.go）会向ErrorLogger输出 panic 信息与完整堆栈：

var HandlePanic = func(recovered interface{}, funcName string) { ErrorLogger.Println(fmt.Sprintf("%s panic: %v", funcName, recovered)) ErrorLogger.Println(string(debug.Stack())) }

2. 三种停止语义

轮询过程由checkNoActiveGoroutines（unbounded_executor.go）驱动：遍历activeGoroutines计数表，只要存在计数大于 0 的条目，就向InfoLogger打印"仍等待 goroutine 退出"及对应的启动位置，并返回 false 继续等待。

这里有一个值得注意的协作约定：executor 只负责发出取消信号（context 取消），并不强杀 goroutine。任务必须自己监听ctx.Done()并返回，否则StopAndWaitForever可能无限等待——README 示例中select分支的case <-ctx.Done()正是这一契约的标准实现。

3. 全局单例与退出纪律

GlobalUnboundedExecutor的注释明确告诫：它的生命周期与程序对齐，期望main函数显式调用 stop，它并不能"神奇地感知"主函数退出。换言之，使用全局 executor 的进程必须在退出前调用停止方法，否则后台 goroutine 会随进程一起被"硬终止"，丢失优雅回收的机会。

五、实战组合：一个完整的可运行示例

将 Map 与 Executor 组合使用，可以构造一个"并发写入、统一回收"的完整场景：

package main import ( "context" "fmt" "time" "github.com/modern-go/concurrent" ) func main() { // 1. 线程安全 map：多个 goroutine 并发写入 m := concurrent.NewMap() // 2. 无上限 executor：统一管理所有后台任务 executor := concurrent.NewUnboundedExecutor() for i := 0; i < 10; i++ { key := fmt.Sprintf("worker-%d", i) executor.Go(func(ctx context.Context) { ticker := time.NewTicker(time.Millisecond) defer ticker.Stop() for { select { case <-ctx.Done(): m.Store(key, "stopped") return case <-ticker.C: m.Store(key, time.Now().String()) } } }) } time.Sleep(50 * time.Millisecond) executor.StopAndWaitForever() // 取消所有任务并等待全部退出 // 3. 安全读取结果 if v, ok := m.Load("worker-3"); ok { fmt.Println("worker-3 =>", v) } }

运行方式：在包含本仓库core模块的环境下，进入 core 目录执行go run即可验证。该示例完整覆盖了 README 中提到的三个核心诉求——线程安全共享状态（Map）、统一取消（StopAndWaitForever）、以及任务退出后的状态收敛。

六、使用建议与边界说明

1. 确认协程监听取消信号：Stop只发信号不回收，任务不响应ctx.Done()时StopAndWait*会一直等待，设计后台任务时应始终在select中监听ctx.Done()。
2. 覆盖 HandlePanic 以自定义告警：库提供包级concurrent.HandlePanic与实例级字段两种覆盖入口，生产环境可在此接入指标上报或告警系统，替换默认的日志输出。
3. 低版本 Map 仅保证最小方法集：跨 Go 版本代码请只依赖Load/Store，如需Delete、Range等能力需自行确认目标版本分支的实现。
4. 全局单例须显式停止：GlobalUnboundedExecutor不会自动感知main退出，进程退出前记得调用其停止方法。
5. 日志默认静默：InfoLogger默认写入ioutil.Discard（见 log.go），排查"为何一直等待退出"这类问题时，可将InfoLogger重定向到 stderr 以观察checkNoActiveGoroutines的等待日志。

七、在本仓库中的定位

本仓库中，该库以 vendored 依赖形式存在于 core/vendor/github.com/modern-go/concurrent，并在 core/go.mod 中声明为github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect，即 wandb-core 通过modern-go生态引入的传递性依赖，其Map与Executor抽象为上层并发代码提供了可移植、可取消的并发原语。对于想要理解 Go 并发封装惯用法的读者，这份源码是一份极佳的微型范本：构建标签实现多版本兼容、context 驱动的统一取消、recover 兜底与生命周期计数，三个模式都浓缩在不足两百行的代码中。

• 机器学习
• 深度学习
• 数据可视化
• 可观测性