【仓颉语言入门 · 第27课】

第 27 课 线程同步：互斥锁、原子类型与条件变量

上节课结尾的 CounterBox 惨案还悬着：10 个线程各加 10000 次，期望 100000，实际只有 33017、33874……每次跑数字还都不一样。本课就来算账：用 Mutex 互斥锁把"读—加—写"三步合成一步，用 Atomic 原子类型让自增本身不可分割，再用 Condition 条件变量解决"线程之间你等我、我等你"的通信问题。

另外，本课会先实测一个上节课预告里提到、很多资料也在讲的东西——Channel 通道，并如实告诉你仓颉 SDK 1.2.0 的现状，然后用本课学的原语亲手实现一个阻塞队列（通道最核心的部分）。

本文所有代码与输出均在仓颉 SDK 1.2.0 下逐行实测编译运行。

目录（系列导航）

整套路线共 7 个模块、30 课：

一、先破案：100000 是怎么丢的

先回顾上节课的"罪犯代码"：

class CounterBox {
var count: Int64 = 0
}

// 10 个线程，每个把 box.count += 1 执行 10000 次

count += 1 看起来是"一句话"，但对 CPU 来说是三条指令：

两个线程可能同时读到同一个旧值：

线程 A：读 100 ──→ 加到 101 ──→ 写回 101
线程 B： 读 100 ──→ 加到 101 ──→ 写回 101

两次加法，结果只涨了 1——另一次凭空丢了。10 个线程抢得越凶，丢得越多。

破案结论：只要一个操作中间能被别的线程插进来，它就是不安全的。解法也很直接——想办法把这三步"锁"在一起，让同一时刻只有一个线程能动 count。这就是同步原语干的事。

📌 本课所有类型都在 std.sync 包里（sync = synchronization，同步），用到时文件开头写 import std.sync.*。spawn、Future、get()、sleep、Duration 仍在 std.core，不用 import。

二、Mutex 互斥锁：lock / unlock 与 synchronized 块

Mutex（mutual exclusion 的缩写，读作"缪泰克斯"）就是一把锁。最朴素的用法是手动 lock() / unlock()：

import std.sync.*

main(): Int64 {
let lock = Mutex()
lock.lock()
println("我拿到锁了")
lock.unlock()

if (lock.tryLock()) {
println("这次也拿到了")
lock.unlock()
} else {
println("锁被占着，没拿到")
}
return 0
}

运行结果：

我拿到锁了
这次也拿到了

规则很简单：

• lock()：拿锁。锁如果正被别的线程拿着，我就站在门口等（阻塞），直到拿到为止；
• unlock()：放锁，让给别人；
• tryLock()：试一下，拿到返回 true，拿不到不等、立刻返回 false。适合"拿不到就算了"的场景。

但手动配对容易出错——加了锁忘了解，其他线程就得永远排队。仓颉提供了更省心的 synchronized 块：

import std.sync.*

main(): Int64 {
let lock = Mutex()
var count = 0
synchronized (lock) {
count += 1 // 这段代码同一时刻只有一个线程能执行
}
println("count = ${count}")
return 0
}

count = 1

synchronized (lock) { … } 的意思是：“执行花括号里的代码前先拿锁，花括号结束时自动放锁”。

自动到什么程度？里面抛异常，锁也照放不误：

import std.sync.*

main(): Int64 {
let lock = Mutex()
try {
synchronized (lock) {
throw IllegalArgumentException("故意捣乱")
}
} catch (e: IllegalArgumentException) {
println("捕获异常：${e.message}")
}
lock.lock()
println("锁还能拿到：synchronized 已自动释放")
lock.unlock()
return 0
}

捕获异常：故意捣乱
锁还能拿到：synchronized 已自动释放

📌 结论：99% 的场景都用 synchronized，不要手写 lock/unlock。 本课后面只有"创建条件变量"那一处必须手动加锁（第六节会讲原因）。

🔸 版本小提示：SDK 1.2.0 的 Mutex 本身就是可重入的——同一个线程对同一把锁连续 lock() 两次不会死锁，解两次即可。旧资料里的 ReentrantMutex 已废弃，编译会告警：warning: class 'ReentrantMutex' is deprecated. Use 'public class Mutex' instead.，看到它直接换成 Mutex 即可。

三、修复 CounterBox（方案一：synchronized）

把上节课的 CounterBox 改造一下：给它配一把锁，所有对 count 的修改都必须穿过这把锁：

import std.sync.*
import std.collection.ArrayList

class SafeCounter {
let lock = Mutex()
var count: Int64 = 0

func addOne() {
synchronized (lock) {
count += 1
}
}
}

main(): Int64 {
let box = SafeCounter()
let futures = ArrayList<Future<Unit>>()
for (_ in 1..=10) {
futures.add(spawn {
for (_ in 1..=10000) {
box.addOne()
}
})
}
for (f in futures) {
f.get()
}
println("期望 100000，实际 ${box.count}")
return 0
}

连跑三次：

期望 100000，实际 100000

期望 100000，实际 100000

期望 100000，实际 100000

三次都是 100000。原来的"读—改—写"被 synchronized 包成了一个不可分割的整体：线程 B 想读，必须等线程 A 写完放锁，读到的永远是最新值。账，算平了。

注意两个和上节课一脉相承的细节：

四、Atomic 原子类型

Mutex 是"谁都别抢，排队来"。还有一种更轻量的思路：让变量本身的读写操作天生不可分割——这就是原子类型（atomic，"原子"取"不可再分"之意）。

4.1 有哪些原子类型

std.sync 提供 9 个：

4.2 数值型原子类型的方法

以 AtomicInt64 为例：

import std.sync.*

main(): Int64 {
let a = AtomicInt64(10)
println("fetchAdd 旧值：${a.fetchAdd(5)}")
println("现在：${a.load()}")
println("swap 旧值：${a.swap(100)}")
println("现在：${a.load()}")
println("CAS 成功？${a.compareAndSwap(100, 200)}")
println("现在：${a.load()}")
println("CAS 失败？${a.compareAndSwap(100, 999)}")
println("现在：${a.load()}")
a.fetchSub(50)
println("fetchSub 后：${a.load()}")
return 0
}

逐行实测输出：

fetchAdd 旧值：10
现在：15
swap 旧值：15
现在：100
CAS 成功？true
现在：200
CAS 失败？false
现在：200
fetchSub 后：150

方法清单：

注意两个初学者最容易踩的点：

AtomicBool 用得最多的是 compareAndSwap（简称 CAS）：

import std.sync.*

main(): Int64 {
let b = AtomicBool(false)
println("Bool CAS：${b.compareAndSwap(false, true)}")
println("Bool 值：${b.load()}")
b.store(false)
println("store 后：${b.load()}")
return 0
}

Bool CAS：true
Bool 值：true
store 后：false

五、修复 CounterBox（方案二：AtomicInt64）

计数器只有一个整数在变，用原子类型比加锁更直接：

import std.sync.*
import std.collection.ArrayList

class AtomicCounter {
let count = AtomicInt64(0)

func addOne() {
count.fetchAdd(1)
}
}

main(): Int64 {
let box = AtomicCounter()
let futures = ArrayList<Future<Unit>>()
for (_ in 1..=10) {
futures.add(spawn {
for (_ in 1..=10000) {
box.addOne()
}
})
}
for (f in futures) {
f.get()
}
println("期望 100000，实际 ${box.count.load()}")
return 0
}

期望 100000，实际 100000

fetchAdd(1) 这一个调用就完成了"读—加—写"，中间不可能插进第二个线程，所以连锁都省了。读结果时记得用 load()，不能直接写 box.count。

两种方案怎么选？

• 只保护一个数字（计数器、序号、开关标志）→ 用 Atomic，更轻、更快；
• 要保护一段逻辑或多个字段（比如"x 和 y 必须一起改"、“先判断再修改”）→ 用 Mutex + synchronized。一把锁里想包几行包几行，原子类型管不了多行逻辑。

六、线程间的等待与通知：Condition 条件变量

互斥锁解决了"抢"的问题，但并发里还有一类"等"的问题：消费者线程要等生产者把数据准备好。一直循环问"好了吗？好了吗？"（忙等）既浪费 CPU，又拿着锁不放。

6.1 一个完整的等待—通知

仓颉的解法是 Condition 条件变量：等的人调用 wait() 睡过去，准备好的人调用 notify() 把它叫醒。

import std.sync.*
import std.collection.ArrayList

main(): Int64 {
let lock = Mutex()
lock.lock()
let notEmpty = lock.condition() // 条件变量必须由"持锁线程"创建
lock.unlock()
let queue = ArrayList<Int64>()

// 消费者：队列空就等
let consumer = spawn {
synchronized (lock) {
while (queue.size == 0) {
notEmpty.wait() // 睡：放锁、等人 notify
}
println("消费者拿到：${queue[0]}")
}
}

sleep(Duration.millisecond * 200)

// 生产者：放数据，叫醒消费者
synchronized (lock) {
queue.add(42)
notEmpty.notify()
println("生产者放入：42")
}
consumer.get()
return 0
}

生产者放入：42
消费者拿到：42

执行过程是这样的：

6.2 三个必须记住的规矩

规矩一：条件变量只能由持锁线程创建。

lock.condition() 要求调用时本线程已经拿着这把锁，不然后果是一个运行时异常（就是上面 main 里先 lock() 再创建再 unlock() 的原因）：

IllegalSynchronizationStateException: Mutex is not locked by current thread.

规矩二：wait() 必须放在 while 循环里，不能用 if。

线程从 wait() 醒来时，条件不一定还成立（可能被别的线程抢先改回去了，也可能存在虚假唤醒）。用 while，醒来后会再检查一次条件，不成立就继续睡；用 if 就直接往下冲了。

规矩三：wait() / notify() 都要在 synchronized (lock) 块里调用。

notify() 只叫醒一个等待者；想全叫醒用 notifyAll()。一个生产者配一个消费者时 notify() 就够；多个消费者抢任务时要用 notifyAll()。

6.3 带超时的等待

不想无限等？给 wait 传 timeout: 参数，返回 Bool：被通知唤醒返回 true，超时返回 false。

import std.sync.*

main(): Int64 {
let lock = Mutex()
lock.lock()
let cond = lock.condition()
lock.unlock()
synchronized (lock) {
let r = cond.wait(timeout: Duration.millisecond * 100)
println("超时等待返回：${r}")
}
return 0
}

超时等待返回：false

注意参数名 timeout: 不能省，漏写会报 error: missing argument prefix 'timeout:' for named parameter。

七、Channel 之谜：1.2.0 实测没有，那就自己造一个

7.1 先说结论

很多并发语言（比如 Go）提倡一种模型：“不要通过共享内存来通信，而要通过通信来共享内存”——线程之间不直接抢变量，而是往一个叫 Channel（通道） 的管子里发消息、收消息，数据同一时刻只在一个线程手里，从根上避免数据竞争。

上节课的预告说这节课讲 Channel。备课的时候我们做了一件本系列一贯的事：在仓颉 SDK 1.2.0 里亲手试。结果是：

import std.sync.*

main(): Int64 {
let ch = Channel<Int64>()
ch.send(1)
return 0
}

error: undeclared identifier 'Channel'

截至仓颉 SDK 1.2.0，标准库里还没有公开的 Channel 类型（std.sync 里有锁、条件变量、原子类型、信号量等，但没有通道）。网上不少文章里的 import concurrency、channel.send() 之类写法在 1.2.0 下都编不过，请大家以本机 SDK 的实测为准。

那"通道"这个概念就学不了了吗？恰恰相反——理解了锁和条件变量，你会发现通道一点都不神秘：它就是一个加了锁的队列，配上两个条件变量。 我们现在就亲手实现一个。

7.2 实现一个 BlockingQueue（阻塞队列）

通道最核心的行为有两个：

• 发送：队列满了就等（等消费者腾位置）；
• 接收：队列空了就等（等生产者放数据）。

用"一把锁 + 两个条件变量 + 第 8 课的 ArrayList"实现。注意删除队首元素用的是第 8 课教过的区间写法 buf.remove(0..1)：

import std.sync.*
import std.collection.ArrayList

class BlockingQueue<T> {
let buf = ArrayList<T>()
let capacity: Int64
let lock = Mutex()
let notEmpty: Condition
let notFull: Condition

init(capacity: Int64) {
this.capacity = capacity
lock.lock()
this.notEmpty = lock.condition()
this.notFull = lock.condition()
lock.unlock()
}

func put(item: T) {
synchronized (lock) {
while (buf.size >= capacity) {
notFull.wait() // 队列满：等消费者取走
}
buf.add(item)
notEmpty.notify() // 放入后：叫醒等数据的人
}
}

func take(): T {
synchronized (lock) {
while (buf.size == 0) {
notEmpty.wait() // 队列空：等生产者放入
}
let item = buf[0]
buf.remove(0..1)
notFull.notify() // 取走后：叫醒等位置的人
return item
}
}
}

对照 Channel 的概念看这份实现：

7.3 跑一个生产者—消费者

一个生产者放 5 个数，一个消费者收 5 个数。为了让输出稳定可对照，只让消费者打印：

main(): Int64 {
let q = BlockingQueue<Int64>(2)

let producer = spawn {
for (i in 1..=5) {
q.put(i)
}
}
let consumer = spawn {
for (_ in 1..=5) {
let v = q.take()
println("收到：${v}")
}
}
producer.get()
consumer.get()
println("收完了")
return 0
}

连跑三次，输出都一样：

收到：1
收到：2
收到：3
收到：4
收到：5
收完了

生产者放得快也没用——队列容量只有 2，放满就睡；消费者取走一个才腾出一个位置。虽然两个线程在并发推进，但数据经过队列时严格先进先出，所以消费者收到的顺序恒为 1→5。这就是"用通信来共享内存"的雏形：谁拿到队列里的数据，谁才有权处理它。

🔸 这个类刻意只保留了通道最本质的骨架，没有做关闭（close）、多消费者唤醒等工程细节。等官方 Channel 在后续 SDK 版本落地后，优先用官方实现；但只要理解了这份实现，以后用谁的通道都一样。

八、其它同步原语速览

8.1 Semaphore 信号量：限流

Semaphore(n) 可以理解为一沓共 n 张的"通行证"：acquire() 领一张（发完了就等），release() 还一张。典型用途是限流——比如同时只允许 2 个线程下载：

import std.sync.*

main(): Int64 {
let sem = Semaphore(2)
sem.acquire()
sem.acquire()
println("许可用完时 tryAcquire：${sem.tryAcquire()}")
sem.release()
println("释放一个后 tryAcquire：${sem.tryAcquire()}")
return 0
}

许可用完时 tryAcquire：false
释放一个后 tryAcquire：true

• acquire()：领通行证，没有就等；
• tryAcquire()：试领，没有立刻返回 false；
• release()：还回一张。

多线程用法就是模板：进临界区前 acquire()，出来 release()（务必成对）。课后练习第 4 题会用它统计"同时在场人数的峰值"。

8.2 Barrier 栅栏：等人到齐再一起走

Barrier(n) 是一个集合点：n 个线程都到达（都调用 wait()）之前，大家一起等；最后一个到的瞬间，全体同时放行。像旅游团"人齐了再发车"：

import std.sync.*
import std.collection.ArrayList

main(): Int64 {
let barrier = Barrier(3)
let futures = ArrayList<Future<Unit>>()
for (i in 1..=3) {
futures.add(spawn {
println("线程 ${i} 到达集合点")
barrier.wait()
println("线程 ${i} 一起出发")
})
}
for (f in futures) {
f.get()
}
return 0
}

一次实测输出：

线程 3 到达集合点
线程 1 到达集合点
线程 2 到达集合点
线程 3 一起出发
线程 1 一起出发
线程 2 一起出发

线程编号的先后顺序每次可能不同（老规矩：调度器说了算），但有一条规律铁打不变：三行"到达集合点"一定全部出现之后，才会出现任何一行"一起出发"。这正是栅栏保证的。

8.3 ReadWriteLock 读写锁：读共享、写独占

ReadWriteLock 里有两把锁：readLock（读锁）和 writeLock（写锁）。规则是：多个线程可以同时持读锁，但写锁同一时刻只能一个线程持有，且持写锁时别人连读都不能读。适合"读多写少"的数据（比如配置表）：

import std.sync.*

main(): Int64 {
let rw = ReadWriteLock()
rw.readLock.lock()
println("拿到读锁")
rw.readLock.unlock()
rw.writeLock.lock()
println("拿到写锁")
rw.writeLock.unlock()
return 0
}

拿到读锁
拿到写锁

🔸 旧资料里还可能看到 Monitor 类型（wait/notify 写在对象本身上），它在 1.2.0 已废弃，编译告警：warning: class 'Monitor' is deprecated. Use 'public interface Condition' instead.。新代码统一用第六节的 Mutex + Condition 写法。

九、CIDE 实操：连跑三次，三个 100000

在 CIDE 里 cjpm init –name syncfix 新建工程，把第三节"方案一"的完整代码贴进 src/main.cj，点运行，然后连按三次运行：

期望 100000，实际 100000

期望 100000，实际 100000

期望 100000，实际 100000

三次一模一样。再翻出上节课第 26 课第七节那个没加锁的 CounterBox 对照着跑三次——33017、33874 之类的乱数 vs 三个 100000，这就是"有没有同步"最直观的对比。

再做一个破坏实验（记得改回来）：把 addOne() 里的 synchronized (lock) { … } 删掉花括号里的锁、变回裸 count += 1，运行，数字立刻开始乱跳。加锁的代码哪里都没坏，去掉锁才坏——并发 Bug 不会在你写代码时报错，只会在运行结果里悄悄错。

十、常用 API 速查

十一、常见问题 FAQ

Q1：为什么编译报 error: undeclared identifier 'Mutex'？

没写 import std.sync.*。spawn、sleep 这些在 std.core 不用导，但锁、原子类型、信号量都在 std.sync，必须显式导入。

Q2：synchronized 块和手动 lock/unlock 用哪个？

一律用 synchronized。它在块结束时（包括异常跳出时）自动放锁，不会忘记；手动 unlock 只在创建条件变量这类特殊场景下用。

Q3：lock.condition() 为什么报 IllegalSynchronizationStateException？

条件变量必须由"当前正持锁的线程"创建。照第六节的固定写法：先 lock.lock()，再 let cond = lock.condition()，然后 lock.unlock()，之后把 cond 分享给各线程使用。

Q4：1.2.0 真的没有 Channel 吗？我看网上文章都在用。

以本机 SDK 实测为准：写 Channel<Int64>() 会得到 error: undeclared identifier 'Channel'。1.2.0 的 std.sync 只提供锁、条件变量、原子类型、信号量、栅栏、读写锁等。第七节已经用这些原语实现了通道最核心的阻塞队列；官方 Channel 请关注后续 SDK 版本。

Q5：原子变量为什么不能直接读写 a.value？

原子类型没有公开的 value 字段（报 error: can not access field 'value'）。读用 load()、写用 store()，所有访问都走原子方法，才能保证并发安全。

Q6：wait() 为什么必须写在 while 里，if 不行吗？

线程从 wait() 醒来时条件未必仍成立（可能被其他线程抢先改变，或出现虚假唤醒）。while 会在醒来后重新检查一次条件，不满足就继续等；if 只检查一次，醒来就硬着头皮往下走，容易出错。这是条件变量的标准写法，照抄即可。

十二、课后练习

class Position {
let lock = Mutex()
var x: Int64 = 0
var y: Int64 = 0

func move(step: Int64) {
// TODO：用 synchronized (lock) 把 x、y 同时加上 step
}
}
// main：10 个 spawn，每个循环 1000 次调用 p.move(1)；
// 全部 get() 后打印 println("x = ${p.x}，y = ${p.y}")

import std.sync.*
// func tryInit(flag: AtomicInt64): Bool { return flag.compareAndSwap(0, 1) }
// 提示：main 里 let flag = AtomicInt64(0)，把 flag 传进每个 spawn

等待开门
开门
开始干活

let lock = Mutex()
// TODO：lock.lock() → let cond = lock.condition() → lock.unlock()
let worker = spawn {
synchronized (lock) {
println("等待开门")
// TODO：cond.wait()
println("开始干活")
}
}
sleep(Duration.millisecond * 200)
synchronized (lock) {
println("开门")
// TODO：cond.notify()
}
worker.get()

下节预告

本课把"线程之间如何安全地协作"讲完了：锁防抢、原子类型防丢数、条件变量防忙等、阻塞队列负责传话。第 28 课 并发实战：多线程任务处理 把这些家伙拉到真正的业务场景里遛一遛：一批任务怎么派给多个工作线程并发处理、处理结果怎么按顺序回收、生产过快消费过慢时怎么用阻塞队列削峰填谷。三节课的并发知识会在那一课串成一条线。

系列说明：本系列基于 Windows 平台 + CIDE + 仓颉 SDK（1.2.0）编写，所有代码均已实际编译运行通过。如遇 SDK 版本差异导致的细节出入，以你本地版本为准，欢迎评论区交流。

💬 遇到问题？扫码联系作者

跟着课程练习时，如果在 SDK 安装、环境变量配置、编译报错或调试上卡住，欢迎扫码加作者企业微信直接咨询（请备注"仓颉课程"）：

离线环境下图片可能加载不出来，也可以在 CIDE 菜单 Help ▸ 联系作者 / Contact 中查看同一张二维码（应用内置兜底图，无需联网）。

📥 工具下载

本系列全程使用的仓颉 IDE —— CIDE（免费开源、社区版）：

• GitCode 仓库 / 安装包下载：https://gitcode.com/wp_upala/cide
• 打开页面后进入 发行版（Releases），两种包任选其一：
  • 安装版：下载 CIDE-<版本>-x64-Setup.exe，双击安装，适合日常长期使用；
  • 免安装版（Portable）：下载 CIDE-<版本>-x64-Portable.zip，解压到任意目录即用，不写注册表、不留安装痕迹，拷到 U 盘也能在别的电脑直接运行（包内附《使用说明.txt》）。适合先试用、或在受限电脑上学习本系列课程。
• 仓颉 SDK 请前往仓颉编程语言官网下载：https://cangjie-lang.cn