第 27 课 线程同步互斥锁、原子类型与条件变量上节课结尾的CounterBox惨案还悬着10 个线程各加 10000 次期望 100000实际只有 33017、33874……每次跑数字还都不一样。本课就来算账用Mutex互斥锁把读—加—写三步合成一步用Atomic原子类型让自增本身不可分割再用Condition条件变量解决线程之间你等我、我等你的通信问题。另外本课会先实测一个上节课预告里提到、很多资料也在讲的东西——Channel 通道并如实告诉你仓颉 SDK 1.2.0 的现状然后用本课学的原语亲手实现一个阻塞队列通道最核心的部分。本文所有代码与输出均在仓颉 SDK 1.2.0 下逐行实测编译运行。目录系列导航整套路线共7 个模块、30 课模块课次内容一、环境与入门0105环境搭建与 Hello World、变量与基本类型、运算符与输入输出、分支、循环二、常用类型与数据组织0610字符串、数组与区间、ArrayList/HashMap/HashSet、可空类型、错误处理三、函数与函数式1114函数、Lambda 与高阶函数、闭包、迭代器与惰性序列四、面向对象与类型系统1520struct/class、构造与属性、接口、枚举与 match 模式匹配、泛型、扩展五、工程化与标准库2125cjpm 包管理与多文件、文件 IO、JSON 处理、网络编程、单元测试六、并发编程2628线程的创建与等待、线程同步本文、并发实战七、项目实战2930命令行小工具、GeoJSON 数据处理实战环境搭建与第一个仓颉程序变量、常量与基本数据类型运算符与标准输入输出分支结构与 match 表达式循环结构while / for / Range字符串详解与字符串插值数组 Array 与区间 Range集合框架ArrayList、HashMap、HashSet可空类型?与 Option错误处理异常机制与 Result函数定义、参数与返回值Lambda 与高阶函数闭包、作用域与函数类型迭代器 Iterator 与 Sequence结构体 struct 与类 class构造函数、属性与方法接口 interface 与实现枚举 enum、代数数据类型与 match 模式匹配泛型编程扩展、类型别名与可见性控制cjpm 包管理与多文件项目组织文件与目录 IOJSON 处理网络编程入门单元测试并发基础线程的创建与等待线程同步互斥锁、原子类型与条件变量本文并发实战多线程任务处理实战一带文件持久化的命令行小工具实战二GeoJSON 数据处理程序一、先破案100000 是怎么丢的先回顾上节课的罪犯代码class CounterBox { var count: Int64 0 } // 10 个线程每个把 box.count 1 执行 10000 次count 1看起来是一句话但对 CPU 来说是三条指令读把内存里count的当前值读到寄存器改寄存器里的值加 1写把新值写回内存。两个线程可能同时读到同一个旧值线程 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 块Mutexmutual 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 1synchronized (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 ArrayListFutureUnit() 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 写完放锁读到的永远是最新值。账算平了。注意两个和上节课一脉相承的细节box仍然是let引用不变可变的是对象内部spawn 的捕获规则没变最后必须逐个get()等齐 10 个线程否则main一结束进程就收队了。四、Atomic 原子类型Mutex 是谁都别抢排队来。还有一种更轻量的思路让变量本身的读写操作天生不可分割——这就是原子类型atomic原子取不可再分之意。4.1 有哪些原子类型std.sync提供 9 个类型包装的普通类型AtomicBoolBoolAtomicInt8/AtomicInt16/AtomicInt32/AtomicInt64Int8/Int16/Int32/Int64AtomicUInt8/AtomicUInt16/AtomicUInt32/AtomicUInt64UInt8/UInt16/UInt32/UInt644.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方法清单方法作用返回值load()读当前值当前值store(v)写入新值无swap(v)换成新值旧值fetchAdd(v)/fetchSub(v)加 / 减旧值fetchAnd(v)/fetchOr(v)/fetchXor(v)位与 / 位或 / 位异或旧值compareAndSwap(expect, new)当前值等于expect才换成new是否交换成功Bool注意两个初学者最容易踩的点fetchAdd返回的是旧值不是加完的值——上面输出里fetchAdd(5)打印 10但接着load()是 15原子类型没有.value字段读值必须用load()。直接写a.value编译器会报error: can not access field value。AtomicBool用得最多的是compareAndSwap简称 CASimport 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 CAStrue 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 ArrayListFutureUnit() 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实际 100000fetchAdd(1)这一个调用就完成了读—加—写中间不可能插进第二个线程所以连锁都省了。读结果时记得用load()不能直接写box.count。两种方案怎么选只保护一个数字计数器、序号、开关标志→ 用Atomic更轻、更快要保护一段逻辑或多个字段比如x 和 y 必须一起改、“先判断再修改”→ 用Mutexsynchronized。一把锁里想包几行包几行原子类型管不了多行逻辑。六、线程间的等待与通知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 ArrayListInt64() // 消费者队列空就等 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执行过程是这样的消费者先进synchronized发现队列是空的执行wait()——睡过去同时把锁交出来不然生产者永远进不来就死锁了200 毫秒后生产者拿到锁、放入 42、notify()叫醒消费者消费者重新拿到锁从wait()醒来再看一眼while条件——现在不空了退出循环取数据。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 ChannelInt64() 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 BlockingQueueT { let buf ArrayListT() 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 的概念看这份实现Channel 概念本实现的对应物发送send/ 接收receiveput/take带缓冲通道容量 nBlockingQueueT(n)发送时缓冲满则阻塞put里的notFull.wait()接收时缓冲空则阻塞take里的notEmpty.wait()无缓冲通道发送方和接收方当面交接容量传1即可近似7.3 跑一个生产者—消费者一个生产者放 5 个数一个消费者收 5 个数。为了让输出稳定可对照只让消费者打印main(): Int64 { let q BlockingQueueInt64(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 }许可用完时 tryAcquirefalse 释放一个后 tryAcquiretrueacquire()领通行证没有就等tryAcquire()试领没有立刻返回falserelease()还回一张。多线程用法就是模板进临界区前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 ArrayListFutureUnit() 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 速查功能写法备注导入同步包import std.sync.*本课类型全在std.sync互斥锁Mutex()1.2.0 可重入ReentrantMutex已废弃同步块synchronized (lock) { ... }块结束自动放锁异常也放手动加/放锁lock.lock()/lock.unlock()必须成对优先用 synchronized尝试加锁lock.tryLock()返回Bool拿不到不等创建条件变量lock.condition()调用时必须已持锁类型写Condition等待cond.wait()必须在 synchronized 里且放在 while 中限时等待cond.wait(timeout: Duration.second * 1)通知返回true超时返回false通知cond.notify()/cond.notifyAll()唤醒一个 / 全部等待者原子整数AtomicInt64(0)另有 8/16/32 位及无符号版本原子布尔AtomicBool(false)load()/store()/compareAndSwap()原子自增a.fetchAdd(1)返回旧值读新值用load()比较交换a.compareAndSwap(expect, new)相等才换返回是否成功信号量Semaphore(n)acquire()/tryAcquire()/release()栅栏Barrier(n)n 个线程都wait()后一起放行读写锁ReadWriteLock().readLock.lock()、.writeLock.lock()配对 unlock十一、常见问题 FAQQ1为什么编译报error: undeclared identifier Mutex没写import std.sync.*。spawn、sleep这些在std.core不用导但锁、原子类型、信号量都在std.sync必须显式导入。Q2synchronized 块和手动 lock/unlock 用哪个一律用synchronized。它在块结束时包括异常跳出时自动放锁不会忘记手动 unlock 只在创建条件变量这类特殊场景下用。Q3lock.condition()为什么报 IllegalSynchronizationStateException条件变量必须由当前正持锁的线程创建。照第六节的固定写法先lock.lock()再let cond lock.condition()然后lock.unlock()之后把cond分享给各线程使用。Q41.2.0 真的没有 Channel 吗我看网上文章都在用。以本机 SDK 实测为准写ChannelInt64()会得到error: undeclared identifier Channel。1.2.0 的std.sync只提供锁、条件变量、原子类型、信号量、栅栏、读写锁等。第七节已经用这些原语实现了通道最核心的阻塞队列官方 Channel 请关注后续 SDK 版本。Q5原子变量为什么不能直接读写a.value原子类型没有公开的value字段报error: can not access field value。读用load()、写用store()所有访问都走原子方法才能保证并发安全。Q6wait() 为什么必须写在 while 里if 不行吗线程从wait()醒来时条件未必仍成立可能被其他线程抢先改变或出现虚假唤醒。while会在醒来后重新检查一次条件不满足就继续等if只检查一次醒来就硬着头皮往下走容易出错。这是条件变量的标准写法照抄即可。十二、课后练习必做用锁保护一个坐标点x、y两个字段必须一起移动这类多字段复合操作原子类型管不了只能用锁。把下面程序的move方法补完整然后起 10 个线程各调用move(1)1000 次。需要import std.sync.*和import std.collection.ArrayList。期望输出x 10000y 10000。class Position { let lock Mutex() var x: Int64 0 var y: Int64 0 func move(step: Int64) { // TODO用 synchronized (lock) 把 x、y 同时加上 step } } // main10 个 spawn每个循环 1000 次调用 p.move(1) // 全部 get() 后打印 println(x ${p.x}y ${p.y})必做用compareAndSwap实现一次性初始化5 个线程同时去抢最多只能有一个成功。函数签名已经给好main 里起 5 个线程进入后先sleep(Duration.millisecond * 100)让大家尽量同时把每个FutureBool的结果在主线程get()回来数true的个数。期望输出成功初始化的线程数1。import std.sync.* // func tryInit(flag: AtomicInt64): Bool { return flag.compareAndSwap(0, 1) } // 提示main 里 let flag AtomicInt64(0)把 flag 传进每个 spawn必做“开门令”工作线程先打印等待开门然后在条件变量上等待主线程睡 200 毫秒后打印开门并通知它工作线程被唤醒后打印开始干活。请补全下面骨架中的三处 TODO提示创建条件变量的固定写法见第六节等待和通知都要包在synchronized (lock)里。期望输出严格按下面三行顺序出现等待开门 开门 开始干活let lock Mutex() // TODOlock.lock() → let cond lock.condition() → lock.unlock() let worker spawn { synchronized (lock) { println(等待开门) // TODOcond.wait() println(开始干活) } } sleep(Duration.millisecond * 200) synchronized (lock) { println(开门) // TODOcond.notify() } worker.get()选做用Semaphore(2)模拟只有 2 个位的更衣室6 个人线程进场每人进去后停留sleep(Duration.millisecond * 100)再出来。用一个带Mutex的装箱类统计同时在场人数的峰值注意计数变量是var不能直接被 spawn 捕获要像第 26 课那样装进 class。期望输出同时在更衣室的峰值人数2。下节预告本课把线程之间如何安全地协作讲完了锁防抢、原子类型防丢数、条件变量防忙等、阻塞队列负责传话。第 28 课并发实战多线程任务处理把这些家伙拉到真正的业务场景里遛一遛一批任务怎么派给多个工作线程并发处理、处理结果怎么按顺序回收、生产过快消费过慢时怎么用阻塞队列削峰填谷。三节课的并发知识会在那一课串成一条线。系列说明本系列基于 Windows 平台 CIDE 仓颉 SDK1.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