第 28 章 / 共 58 章 卷六 并发

线程与同步原语

第 27 章说"共享是默认"——单线程里这是便利,多线程里这是陷阱:四个线程各加两万次,结果不是八万。本章开卷六:thread 怎么开、mutex/sem/cond/rwlock 四件同步工具怎么用、死锁怎么来又怎么走。所有竞态演示都是真实跑出来的数字。

本章导读

卷五结尾说"共享是默认"在并发世界会有新的分量——本章兑现这个预告。xlang 的并发模型是经典的系统线程 + 共享内存 + 手动同步:thread 直接对应操作系统线程,全局变量、class 实例、容器天然被所有线程看见,语言不替你加锁、不做数据竞争检查。这意味着两件事:威力——线程间通信零拷贝,拿引用就能协作;责任——每一处共享可变状态都是你的。本章先学开车(thread 的创建、传参、等待),再看车祸现场(三种存储介质上竞态的三种后果:丢更新、null 化、段错误——全部实测复现),然后配齐四件安全工具(mutex 互斥、sem 计数、cond 等待唤醒、rwlock 读写分离),最后处理多锁场景的头号事故(死锁)与防御总则。下一章的 channel 会给出"不要共享、要通信"的另一条路线,两章对照着读味道最佳。

本章学习目标

会用 thread(f) 创建线程(入口必须 func() as int)并用闭包捕获/工厂函数传参;掌握 wait/waitTimeout/exitCode/state/stop 五个方法;理解竞态在静态变量、var、闭包捕获块上的三种不同后果(实测数字);会用 mutex 保护临界区(含 tryLock 与非递归特性);会用 sem 做计数同步与限流(post/waitTimeout/tryWait/postMultiple);会用 cond 做等待-唤醒(waitTimeout 须持锁、while 循环防虚假唤醒);会用 rwlock 读写分离(含 downgrade/upgrade);能诊断锁序死锁并用统一锁序修复;记住共享数据的四条防御准则。

28.1 并发模型定位

三句话给卷六定调:

  • thread 是系统线程——真并行,多核同时跑;创建销毁有成本,别开几百个。
  • 内存默认共享——全局变量(静态与 var)、class 实例、容器,所有线程直接可见。第 27 章"赋值共享引用"的规则跨线程依然成立,这正是竞态的温床。
  • 同步全靠手动——语言提供 mutex/sem/cond/rwlock 四件工具,但用不用、用在哪,编译器不管。没有 borrow checker,没有数据竞争告警——28.4 的车祸现场就是不管的代价。

28.2 thread 创建与等待

thread(函数值) 创建并立即启动线程,入口签名固定为 func() as int:

thread.xl
func worker() as int
    return 7
end
thread t = thread(worker)   // 创建即启动
println(t.wait())             // true:阻塞等线程结束
println(t.exitCode())         // 7:返回值在此领取

func slow() as int
    time.sleep(50)
    return 9
end
thread t3 = thread(slow)
println(t3.waitTimeout(0))     // false:还在跑,立刻超时
println(t3.waitTimeout(2000))  // true:两秒内等到了
println(t3.exitCode())         // 9

方法一览(全部实测):wait() 阻塞等待、返回 bool;waitTimeout(ms) 限时等待、超时返回 false;exitCode() 取线程返回值(完成后可反复取,值稳定);state() 返回状态码(int,>=0 正常);stop() 请求停止、返回 bool。入口签名不符会在编译期拦下(坑 1);传空 func 值则是运行期可捕获的异常(坑 2)。

28.3 闭包传参与工厂

入口不能带参数,数据走闭包捕获;每个线程要不同数据时,用工厂函数制造各自捕获的闭包:

capture.xl
int base = 5
auto captured = func() as int
    return base + 2       // 捕获外层局部变量
end
thread t = thread(captured)
println(t.wait())          // true
println(t.exitCode())      // 7

func makeWorker(int code) as func() as int
    return func() as int
        return code          // 每个闭包捕获自己的 code
    end
end
thread a = thread(makeWorker(11))
thread b = thread(makeWorker(12))
a.wait()
b.wait()
println(a.exitCode())      // 11
println(b.exitCode())      // 12

一个重要实测:闭包捕获的变量在线程间是共享的。两个线程跑同一个捕获了局部 x 的闭包、各自增 1000 次,最终 x = 2000 且先结束的线程只读到 1555——捕获块就是一块共享存储,竞态规则对它完全适用(28.4)。想在循环里批量开线程时尤其要走工厂模式,别让所有线程捕获同一个循环变量。

28.4 竞态实录

同一个程序——四个线程各把共享计数器加 20000 次,预期 80000——在三种存储介质上跑出三种后果(全部实测,可复现):

共享介质实测结果后果形态
静态 int 全局变量79135丢更新:读-改-写三步被插队,865 次加法蒸发
全局 varnull值槽击穿:var 装箱值被并发写撕成 null
闭包入口 + 全局 var段错误(退出码 139)进程崩溃:3/3 复现,最惨烈的形态

共享 class 实例字段同样逃不掉:两线程各加 20000 次,40000 的场子实测收 39991。记住这组对照:竞态在 xlang 不只是"数字不对",它能击穿 null、能干掉进程——var 装箱值与闭包捕获块的内部结构都不是并发安全的。防御见 28.5 与 28.10。

28.5 mutex 互斥锁

给临界区上锁,竞态即刻消失——同样的 4×20000,有锁版本精确得 80000:

mutex.xl
int counter = 0
mutex m = mutex()
func racer() as int
    for i in range(20000)
        m.lock()
        counter = counter + 1
        m.unlock()
    end
    return 0
end
thread a = thread(racer)
thread b = thread(racer)
thread c = thread(racer)
thread d = thread(racer)
a.wait()
b.wait()
c.wait()
d.wait()
println(counter)     // 80000:一次不差

API 细节:lock()/unlock()/tryLock() 都返回 bool 可断言;mutex m 裸声明与 mutex() 等价;非递归锁——同一线程对已持有的锁再 tryLock 得 false(想试递归加锁就是死锁,28.9)。锁的粒度是性能与安全的跷跷板:本例每次循环都加锁只为演示,实务中尽量"锁内只碰共享数据,锁外做计算"。

28.6 sem 信号量

sem(初值, 上限) 是一把"计数门":post 加一、wait 减一,减到零就排队。两件经典用法——跨线程门闩与限流:

sem.xl
sem gate = sem(0, 1)       // 门闩:初始关闭
func waiter() as int
    if not gate.waitTimeout(2000) then
        return 10            // 超时兜底
    end
    return 23
end
thread t = thread(waiter)
println(t.waitTimeout(1))   // false:worker 还在等门开
gate.post()                  // 开门
println(t.waitTimeout(2000)) // true
println(t.exitCode())       // 23

API 细节:post()/postMultiple(n)/waitTimeout(ms)/tryWait() 均返回 bool;post 到上限会失败(sem(1,1) 已有一个再 post 得 false),postMultiple 溢出整体失败不部分提交——计数永远不超上限。用 sem(N, N) 限流:N 个许可发完,后来的线程自然排队(练习 2 实测峰值恰为 2)。

28.7 cond 条件变量

sem 数"还有几个",cond 等"某件事发生"。等待方必须持锁调用 waitTimeout(它会在等待期间释放锁、醒来时重新持有):

cond.xl
bool ready = false
mutex m = mutex()
cond c = cond()
func waiter() as int
    m.lock()
    int rounds = 0
    while not ready          // 必须循环复检,防虚假唤醒/超时
        c.waitTimeout(m, 100)
        rounds = rounds + 1
    end
    m.unlock()
    return rounds
end
thread w = thread(waiter)
time.sleep(30)
m.lock()
ready = true
c.signal()                   // 唤醒一个等待者
m.unlock()
w.wait()
println(w.exitCode() > 0)   // true:醒过至少一轮

cond 三律:等待必持锁(waitTimeout(mutex, ms) 签名强制);条件变量配条件谓词(ready 这样的 bool 标记,signal/broadcast 只负责"叫醒",不负责"传话");醒后必复检(while 循环——超时返回 false 也是"醒来",不复检会把超时当信号,坑 5)。多等待者场景用 broadcast() 全体唤醒。

28.8 rwlock 读写锁

读多写少的场景,mutex 太粗——rwlock 让读锁共享、写锁独占:

rwlock.xl
int shared = 0
rwlock rw = rwlock()
func reader() as int
    int total = 0
    for i in range(5)
        rw.readLock()          // 读锁:多读者共存
        total = total + shared
        rw.readUnlock()
        time.sleep(1)
    end
    return total
end
func writer() as int
    for i in range(5)
        rw.writeLock()         // 写锁:独占
        shared = shared + 10
        rw.writeUnlock()
        time.sleep(1)
    end
    return 0
end
thread w = thread(writer)
thread r1 = thread(reader)
thread r2 = thread(reader)
w.wait()
r1.wait()
r2.wait()
println(shared)              // 50
rw.writeLock()
rw.downgrade()               // 写锁降级为读锁,原子完成
println(shared)              // 50:降级后读自己刚写的值
rw.readUnlock()

API 细节:readLock/readUnlock/writeLock/writeUnlock/tryReadLock/tryWriteLock 全返回 bool;downgrade() 写转读(保住"刚写完的数据不被人插改");upgrade() 读转写——未持读锁调用得 false,持读锁升级得 true(升级后配 writeUnlock)。

28.9 死锁与锁序

两把锁、两个线程、相反顺序——教科书死锁,实测重现:

deadlock.xl(请勿模仿)
mutex ma = mutex()
mutex mb = mutex()
func ab() as int
    ma.lock()
    time.sleep(20)      // 放大事故窗口
    mb.lock()            // 等 ba 手里的 mb
    mb.unlock()
    ma.unlock()
    return 1
end
func ba() as int
    mb.lock()
    time.sleep(20)
    ma.lock()            // 等 ab 手里的 ma——互相等待
    ma.unlock()
    mb.unlock()
    return 2
end
thread t1 = thread(ab)
thread t2 = thread(ba)
println(t1.waitTimeout(1000))  // false:一秒了还没完
println(t2.waitTimeout(1000))  // false:两个都僵死,进程无法退出

解法是最古老的那条:统一锁序——所有线程都按 ma→mb 的顺序加锁,循环等待无法形成。把 ba 的加锁顺序改成 ma 在前,程序立刻干净跑完(exitCode 之和为 3)。工程建议:项目里给锁起名字时带上序号(m1_config、m2_cache),代码里永远按序号升序加锁——锁序写在名字里,比写在文档里可靠。

28.10 共享防御准则

把本章的教训压缩成四条:

  1. 共享数据优先静态类型——静态 int 竞态丢更新(数字难看但进程活着),var 竞态击穿 null、配闭包甚至段错误。并发场景里的共享状态,var 能不用就不用。
  2. 锁与数据同生共死——每一处共享可变状态,写下它的同时就写下保护它的锁;review 时按"这行代码碰的共享数据拿的是哪把锁"逐行过。
  3. 锁内只碰共享数据——计算、IO、sleep 放锁外,锁越薄死锁与性能问题越少。
  4. 多锁必定序——28.9 的统一锁序,别无他法。

下一章的 channel 提供了第五条路线的预告:能不共享就不共享——数据在消息里流动,锁的需求自然消失。两条路线都掌握,按场景选型。

实战:四线程数素数

把 thread、工厂闭包、mutex 汇总成一件真事:四个线程分区间统计十万以内素数:

primes.xl
int total = 0
mutex m = mutex()
func isPrime(int n) as bool
    if n < 2 then
        return false
    end
    int d = 2
    while d * d <= n
        if n % d == 0 then
            return false
        end
        d = d + 1
    end
    return true
end
func makeCounter(int lo, int hi) as func() as int
    return func() as int
        int found = 0
        int n = lo
        while n < hi
            if isPrime(n) then
                found = found + 1
            end
            n = n + 1
        end
        m.lock()               // 锁内只有一句共享写
        total = total + found
        m.unlock()
        return found
    end
end
thread t1 = thread(makeCounter(2, 25000))
thread t2 = thread(makeCounter(25000, 50000))
thread t3 = thread(makeCounter(50000, 75000))
thread t4 = thread(makeCounter(75000, 100000))
t1.wait()
t2.wait()
t3.wait()
t4.wait()
println(t1.exitCode() + t2.exitCode() + t3.exitCode() + t4.exitCode())  // 9592
println(total)              // 9592:两种汇总口径一致

三个设计点都值得带走:工厂闭包给每个线程独立的区间(28.3);计算在锁外——isPrime 的全部耗时都在锁外跑,锁内只剩一句 total 累加(28.10 准则 3);双口径验证——exitCode 之和与 mutex 保护的 total 互相印证,少一个都可能是错的。全程 0.17 秒跑完,真并行的甜头。

深入细节

拾遗 1:竞态的三张面孔——为什么后果因介质而异

28.4 的对照表值得多想一层:同样是"四个线程加一个数",静态 int 丢更新、var 变 null、闭包捕获路径直接段错误。原因在三种存储的内部结构:静态 int 槽就是一个机器字,并发写的是值本身,最坏只是读-改-写窗口里丢几次更新;var 槽存的是装箱 xvalue 的指针/标签组合,读-改-写涉及解包、新建装箱值、写回指针多步,两个线程交错执行可能把"半新半旧"的组合写进槽里——null 只是其中一种畸形产物;闭包捕获块还叠加了一层捕获环境对象的内存管理,并发写触发的是运行时内部的引用计数/结构竞争,直接越过"数据错误"升级为"进程崩溃"。教训因此分层:静态类型的竞态是逻辑 bug(难查但可存活),var 与闭包路径的竞态是稳定性事故(难查且致命)。这也是 28.10 准则 1"共享数据优先静态类型"的全部理由。

拾遗 2:全 bool 返回值的同步 API 哲学

注意本章所有同步原语的操作都返回 bool——lock()、unlock()、tryLock()、post()、waitTimeout()、signal()、downgrade(),无一例外。这不是随意风格,是把"同步操作可能失败"写进了类型:满 sem 的 post 会失败、超时的 wait 会失败、无锁可升的 upgrade 会失败。配合 assert(第 13 章),你可以把每一次同步操作都变成可验证的断言点——官方测试用例就是这么写的,满屏 assert m.lock()。实务建议:关键路径上的同步调用要么检查返回值、要么 assert 掉,裸调用等于假设世界永远如你所愿;而 waitTimeout 族返回 false 时(超时)的处理分支,正是给"对方失约了"准备的 Plan B——分布式思维的萌芽。

拾遗 3:从共享内存到消息传递——卷六的路线之争

本章的模型有个名字:共享内存并发。它的威力(零拷贝、表达力)与代价(竞态、死锁、锁序纪律)你都已见过。下一章 channel 代表另一条路线:消息传递并发——线程不共享数据,而是通过管道投递值的所有权,"不要通过共享内存来通信,要通过通信来共享内存"。两条路线不是替代关系:共享内存适合细粒度、高频、读多的协作(本章实战的计数器);消息传递适合粗粒度、流水线、生产者消费者(第 29 章)。第 30 章的 async/await 与 future 则是第三条——协作式并发,单线程内交错。读完卷六回来看,你会发现 xlang 把三条路线都备齐了,选型本身就是架构决策。

⚠️ 常见错误

以下报错全部来自真实编译器输出(文件路径已缩短)。

坑 1:带参函数作线程入口
# bad.xl 内容:func withParam(int x) as int # thread t = thread(withParam)(第 4 行) > xl run bad.xl bad.xl:4:19: [sem] thread entry must match func() as int

原因:线程入口签名锁死为 func() as int,多一个参数都不行。正确做法:参数走闭包捕获(28.3),批量开线程用工厂函数。

坑 2:空 func 值创建线程
# bad.xl 内容:func() as int missing(未赋值) # thread t = thread(missing) > xl run bad.xl (抛出异常:thread requires a callable func value)

原因:声明了 func 变量但没赋值就传给 thread——运行期异常,可 try/catch 捕获(官方用例按消息文本匹配)。正确做法:传参前确认函数值非空;动态选择入口时用 if 判空。

坑 3:无锁累加共享计数器
# 4 线程 × 20000 次 counter = counter + 1 > xl run bad.xl 79135(预期 80000;var 版本输出 null;闭包+var 版本段错误)

原因:自增是读-改-写三步,并发交错丢更新;var 与闭包路径后果升级为 null 化与崩溃。正确做法:临界区上 mutex(28.5);或学实战——线程内先算局部和,最后一次性加锁汇总。

坑 4:递归加同一把 mutex
# m.lock() 后同线程再次 m.lock() # (线程自我僵死;tryLock 版本返回 false)

原因:xlang 的 mutex 非递归——同线程二次加锁不会"计数放行",而是直接互斥。正确做法:加锁前确认调用链上没有持锁者;需要可重入语义时自己封装计数包装。

坑 5:把 cond 超时当成被唤醒
# c.waitTimeout(m, 100) 后直接往下走 # ——返回 false(超时)也是"醒来",谓词可能根本没变

原因:waitTimeout 返回 false 表示超时而非条件满足;signal 也可能在谓词变化前到达。正确做法:while not 谓词 循环包裹等待(28.7 标准模式),醒后复检再行动。

本章小结

主题规则要点
threadthread(f) 创建即启动;入口 func() as intwait/waitTimeout/exitCode/state/stop 全 bool 或 int
传参闭包捕获;批量开线程用工厂函数捕获变量线程间共享,竞态规则适用
竞态静态 int 丢更新;var 击穿 null;闭包+var 段错误共享数据优先静态类型
mutexlock/unlock/tryLock 返回 bool非递归;裸声明可用;锁内只碰共享数据
semsem(初值, 上限);post/waitTimeout/tryWait满时 post 失败;限流用 sem(N, N)
condwaitTimeout(mutex, ms) 须持锁while 谓词循环复检;signal 一 broadcast 全
rwlock读共享写独占;downgrade 写转读;upgrade 读转写读多写少首选;upgrade 须先持读锁
死锁相反锁序互等唯一解法:统一锁序(锁名带序号)
模型系统线程+共享内存+手动同步下一章 channel 是消息传递路线

复习题

✍️ 先遮住答案想一想
  1. thread 入口的签名要求是什么?要给线程传三个参数怎么办?
  2. waitTimeout(0) 返回 false 说明什么?exitCode() 在线程结束后能取几次?
  3. 同样的无锁计数程序,静态 int、var、闭包+var 三种写法的实测后果分别是什么?
  4. sem(0, 2) 的 0 和 2 各是什么含义?post 到上限会怎样?
  5. cond.waitTimeout 为什么必须在 while 循环里调用?
  6. 两个线程两把锁的死锁如何形成?唯一可靠的预防是什么?
参考答案
  1. 必须 func() as int,带参报 [sem] thread entry must match func() as int。传参用闭包捕获;每个线程不同参数用工厂函数制造各自捕获的闭包。
  2. 线程仍在运行(立刻超时)。exitCode() 完成后可反复取,值稳定。
  3. 静态 int:丢更新(79135/80000);var:值槽击穿成 null;闭包+var:段错误(退出码 139,稳定复现)。
  4. 0 是初始计数(门闩初始关闭),2 是上限。计数达上限后 post 返回 false 失败,postMultiple 溢出整体失败。
  5. 因为超时(返回 false)与虚假唤醒都会让调用返回,而谓词未必满足;while 循环保证"条件真满足才继续"。
  6. t1 持 ma 等 mb、t2 持 mb 等 ma——循环等待。预防:所有线程统一加锁顺序(如按锁名序号升序)。

编程练习

练习 1:账户加减守恒

编写 bank.xl:全局 balance = 100000,两个线程分别做 10000 次 -1 与 +1,mutex 保护。验证最终余额分毫不差;再把锁去掉跑一次,观察余额漂移。

参考思路:有锁版精确 100000——收支守恒是互斥最直观的体检;无锁版结果不确定(丢更新),两次运行多半不同。对照 28.4 理解漂移来源。

练习 2:信号量限流器

编写 limit.xl:sem(2, 2) 作许可池,四个"工作线程"各自:取许可 → 上报在岗计数(mutex 保护 peak)→ sleep → 离岗 → 还许可。打印峰值并发数。

参考思路:峰值恒为 2——许可发完后来的自然排队;peak 的统计本身也是共享写,别忘 mutex(锁内只读写字段,sleep 在锁外)。

练习 3:读写锁缓存

编写 cache.xl:var dict 当缓存,一个写线程 5 次递增 hits,两个读线程各读 10 次累计,rwlock 保护。打印最终 hits。再把 rwlock 全去掉跑一次对比。

参考思路:有锁版 hits 恒为 5;注意 var 读出参与 int 运算要 int() 转换([sem] 赋值表达式类型和目标变量类型不匹配);无锁版可能见到 null 或更小读数——var 在并发下的脆弱性(28.4)再次现身。