仓颉并发编程实战：std.sync 原子操作、锁、条件变量与定时器使用详解

本篇技术指南围绕仓颉（Cangjie）标准库 std.sync 包中的五类并发工具——Atomic 原子操作、Mutex 可重入互斥锁、Condition 条件变量、Monitor（已废弃）与 Timer 定时器，以官方示例文档为主线，结合 cangjie_runtime 仓库中 stdlib/libs/std/sync 的源码实现与 runtime/src/Sync/Sync.cpp 的运行时原语，逐一给出可复制运行的完整代码、运行结果与底层原理剖析。读完本文，你将能够在多线程仓颉程序中正确使用原子计数、互斥锁保护临界区、条件变量实现线程挂起/唤醒，以及用定时器调度一次性和重复性任务，并理解它们各自的使用约束与弃用迁移路径。

一、认识 std.sync：仓颉并发编程的工具箱

在深入示例之前，先理解 std.sync 包在仓颉并发模型中的定位。根据 stdlib/doc/libs/std_en/sync/sync_package_overview.md 的说明，仓颉语言屏蔽了"操作系统线程"与"用户态线程"的概念差异，统一提供 仓颉线程（Cangjie thread） 这一概念。仓颉线程采用 M:N 线程模型——M 个仓颉线程调度运行在 N 个操作系统线程之上，它们是支持抢占的轻量级用户态线程，相比 OS 线程占用更少的内存资源。

创建仓颉线程只需使用 spawn 关键字后跟一个无参 lambda 表达式：

main () {
    spawn {
        // 在新线程中执行
        println("Thread: ${Thread.currentThread.id}")
    }
    // 在主线程中执行
    println("Thread: ${Thread.currentThread.id}")
    sleep(Duration.second)
    0
}

std.sync 包为并发编程提供了几大类能力：

• 原子操作：支持整型（Int8/Int16/Int32/Int64、UInt8/UInt16/UInt32/UInt64）、Bool 与引用类型（AtomicReference、AtomicOptionReference）；
• 锁：Mutex（可重入互斥锁）、ReadWriteLock（可重入读写锁）、Semaphore（信号量）等；
• 条件变量：Condition 接口，用于线程的挂起与唤醒；
• 定时器：Timer，用于一次性或周期性任务调度。

此外，仓颉还提供 synchronized 关键字与 Lock 接口 配合使用：进入 synchronized 块前自动获取锁（拿不到锁则阻塞），退出块时（包括通过 break、continue、return、throw 等控制转移退出）自动释放锁，从而避免"忘记解锁"或"持锁抛异常导致锁无法释放"的问题。从 stdlib/libs/std/sync/lock_interface.cj 的注释可以看到其语义等价于 try { m.lock(); foo() } finally { m.unlock() }。

下面依次展开官方示例文档 stdlib/doc/libs/std_en/sync/sync_samples/sync_samples.md 中的五类用法。

二、Atomic 原子操作：多线程计数器的正确打开方式

2.1 完整示例：1000 个线程并发累加

在多线程程序中，对共享计数器的"读-改-写"过程如果不用原子操作保护，会出现竞态导致计数丢失。std.sync 的原子类型通过 load（读）、store（写）、swap（交换）、compareAndSwap（CAS）以及 fetchAdd/fetchSub（算术）等操作保证线程安全。示例使用 AtomicInt64 实现一个被 1000 个线程并发累加的计数器：

import std.sync.*
import std.time.*
import std.collection.*

let count = AtomicInt64(0)

main(): Int64 {
    let list = ArrayList<Future<Int64>>()

    /* Create 1000 threads */
    for (_ in 0..1000) {
        let fut = spawn {
            sleep(Duration.millisecond) /* Sleep for 1 millisecond */
            count.fetchAdd(1)
        }
        list.add(fut)
    }

    /* Wait for all threads to complete */
    for (f in list) {
        f.get()
    }

    var val = count.load()
    println("count = ${val}")
    return 0
}

运行结果：

count = 1000

关键点解读：

• count.fetchAdd(1) 是原子自增：它在硬件层面完成"读取当前值 → 加 1 → 写回"这一不可分割的读-改-写序列，返回的是修改前的值（见 sync_package_overview.md 的说明）。正因为每个线程的 fetchAdd 是原子的，1000 个线程最终累加结果严格等于 1000。
• 每个 spawn 返回一个 Future，主线程通过 f.get() 等待所有子线程完成，再 count.load() 读取最终值——load 是原子读操作。
• 示例中先 sleep(Duration.millisecond) 再累加，是为了模拟"各线程并发到达"的场景，验证原子操作在真实并发下的正确性。

2.2 源码剖析：原子类型与编译器内置函数

从 stdlib/libs/std/sync/atomic.cj 可以看到，所有原子类型（AtomicInt8~AtomicInt64、AtomicUInt8~AtomicUInt64、AtomicBool、AtomicReference<T>、AtomicOptionReference<T>）的底层都委托给一组标有 @Intrinsic 注解的编译器内置函数：

@Intrinsic
func load<ATOMIC_TYPE, TYPE>(atomicType: ATOMIC_TYPE, memoryOrder: MemoryOrder): TYPE

@Intrinsic
func store<ATOMIC_TYPE, TYPE>(atomicType: ATOMIC_TYPE, value: TYPE, memoryOrder: MemoryOrder): Unit

@Intrinsic
func swap<ATOMIC_TYPE, TYPE>(atomicType: ATOMIC_TYPE, value: TYPE, memoryOrder: MemoryOrder): TYPE

@Intrinsic
func compareAndSwap<ATOMIC_TYPE, TYPE>(
    atomicType: ATOMIC_TYPE, oldValue: TYPE, newValue: TYPE,
    successOrder: MemoryOrder, failureOrder: MemoryOrder
): Bool

@Intrinsic
func fetchAdd<ATOMIC_TYPE, TYPE>(atomicType: ATOMIC_TYPE, value: TYPE, memoryOrder: MemoryOrder): TYPE

这些 @Intrinsic 函数由编译器直接映射为底层原子指令，因此 fetchAdd、compareAndSwap 等调用没有额外的库函数开销。整型原子类型提供完整的操作集合：

操作	说明
load()	原子读取当前值（默认内存序）
store(val)	原子写入新值
swap(val)	写入新值并返回写入前的旧值
compareAndSwap(old, new)	若当前值等于 old 则写入 new 并返回 true，否则不写入返回 false
fetchAdd(val) / fetchSub(val)	原子加/减，返回修改前的值
fetchAnd(val) / fetchOr(val) / fetchXor(val)	原子位运算（与/或/异或），返回修改前的值

而 Bool 与引用类型（AtomicBool、AtomicReference<T>、AtomicOptionReference<T>）只支持读、写与交换操作（load/store/swap/compareAndSwap），不支持算术与位运算。

需要特别注意的是内存序 API 的弃用：atomic.cj 中每个操作都提供了带 memoryOrder 命名参数的重载版本（例如 load(memoryOrder!: MemoryOrder)），但均标注了 @Deprecated，官方建议改用无参版本，默认使用 MemoryOrder.SeqCst（顺序一致）。原因从源码注释可以看出：带默认参数版本需要额外构造默认参数，相比直接调用无参方法会带来明显的性能开销，因此拆分为两个方法。

此外，AtomicReference<T> 对泛型参数 T 有约束 where T <: Object，即只能原子封装 Object 的子类引用；AtomicOptionReference<T> 则是可空引用（Option<T>）的原子封装，其无参 init() 构造一个空实例。

三、Mutex 可重入互斥锁：保护临界区

3.1 完整示例：Mutex 加锁/解锁与条件等待

当临界区代码较长或需要跨线程协作时，除了原子变量，更通用的方式是互斥锁。Mutex 是可重入互斥锁：同一线程可以多次获取同一把锁，且只有完全释放（加锁 N 次需解锁 N 次）后其他线程才能获取。示例中，子线程在持锁状态下循环 wait()，主线程修改 flag 后调用 notifyAll() 唤醒它：

import std.sync.*

var mt = Mutex()
var con = synchronized(mt) { mt.condition() }
var flag: Bool = true

main(): Int64 {
    let fut = spawn {
        mt.lock()
        while (flag) {
            println("New thread: before wait")
            con.wait()
            println("New thread: after wait")
        }
        mt.unlock()
    }
    /* Sleep for 10 milliseconds to ensure the new thread can execute */
    sleep(10 * Duration.millisecond)
    mt.lock()
    println("Main thread: set flag")
    flag = false
    mt.unlock()

    println("Main thread: notify")
    mt.lock()
    con.notifyAll()
    mt.unlock()

    /* Wait for the new thread to complete */
    fut.get()
    return 0
}

运行结果：

New thread: before wait
Main thread: set flag
Main thread: notify
New thread: after wait

要点解读：

• var con = synchronized(mt) { mt.condition() }：mt.condition() 用于为这把 Mutex 生成一个关联的 Condition 实例。必须在持有锁的状态下调用，否则抛出 IllegalSynchronizationStateException（见下文源码）。
• 子线程 con.wait() 会挂起当前线程，同时释放关联的互斥锁；被唤醒后重新获取锁再继续执行（这是 Condition.wait 的语义，详见 sync_package_interfaces.md）。
• notifyAll() 唤醒所有等待在该条件上的线程；若只想唤醒一个，可用 notify()。
• 主线程先 sleep(10 * Duration.millisecond) 确保子线程先进入 wait()，避免"先 notify 后 wait"导致信号丢失。

3.2 源码剖析：Mutex 的可重入与状态校验

stdlib/libs/std/sync/mutex_cjnative.cj 中 Mutex 的声明为 public class Mutex <: UniqueLock，实现 Lock 接口的 lock()/tryLock()/unlock()，并额外提供 condition()：

public func lock(): Unit {
    mutexLock(this)
}

public func tryLock(): Bool {
    return mutexTryLock(this)
}

public func unlock(): Unit {
    checkMutexStatus()
    mutexUnlock(this)
}

public func condition(): Condition {
    checkMutexStatus()
    return MutexCondition(this)
}

func checkMutexStatus(): Unit {
    let status = mutexCheckStatus(this)
    if (!status) {
        throw IllegalSynchronizationStateException("Cannot unlock mutex: not locked by current thread.")
    }
}

从源码结构可以看出两点设计：

1. 可重入支持：unlock() 前通过 mutexCheckStatus 校验"当前线程必须持有锁"，而 lock() 直接委托运行时 mutexLock，重入计数由运行时维护。相应地，API 文档 sync_package_classes.md 明确说明"若锁被递归加锁 N 次，则需调用 N 次 unlock() 才能完全释放；完全释放后若有其他线程阻塞在该锁上，会唤醒其中一个"。
2. 异常语义：Mutex 构造失败（运行时初始化返回空指针）或"当前线程未持锁却解锁/生成条件"时，都会抛出 IllegalSynchronizationStateException。

在运行时侧，runtime/src/Sync/Sync.h 定义了 struct CJMutex 及 MCC_MutexInit、MCC_MutexLock、MCC_MutexTryLock、MCC_MutexCheckStatus、MCC_MutexUnlock 等底层接口，runtime/src/Sync/Sync.cpp 中的 MCC_MutexLock 调用 MCC_MutexLockImpl，锁不可用时进入 MCC_MutexLockSlowPathImpl 的慢路径阻塞等待。也就是说，std.sync.Mutex 是直接构建在运行时同步原语之上的。

3.3 synchronized：避免手动加解锁的陷阱

如果每个临界区都手动 lock()/unlock()，很容易在异常或提前返回时漏掉 unlock()。仓颉的 synchronized 关键字正是为此设计。以下示例（来自 sync_package_overview.md）展示 5 个线程通过 synchronized(mtx) 串行更新共享计数：

import std.sync.Mutex

main () {
    let mtx = Mutex()
    let cnt = Box<Int64>(0)

    for (_ in 0..5) {
        spawn {
            synchronized(mtx) {
                cnt.value ++
                println("count: ${cnt.value}")
            }
        }
    }
    sleep(Duration.second)
    0
}

可能的输出：

count: 1
count: 2
count: 3
count: 4
count: 5

无论线程以何种顺序进入临界区，synchronized 都保证同一时刻只有一个线程执行块内代码，且退出块（包括异常退出）时锁必定被释放。

四、Monitor（已废弃）：理解历史与迁移路径

4.1 完整示例：Monitor 挂起与唤醒线程

Monitor 是 std.sync 早期的"互斥锁 + 条件变量"一体式实现：它自身同时承担加锁与 wait/notify 的角色。官方文档明确标注其将在未来版本中废弃，示例代码如下：

import std.sync.*

var mon = Monitor()
var flag: Bool = true

main(): Int64 {
    let fut = spawn {
        mon.lock()
        while (flag) {
            println("New thread: before wait")
            mon.wait()
            println("New thread: after wait")
        }
        mon.unlock()
    }

    /* Sleep for 10 milliseconds to ensure the new thread can execute */
    sleep(10 * Duration.millisecond)

    mon.lock()
    println("Main thread: set flag")
    flag = false
    mon.unlock()

    println("Main thread: notify")
    mon.lock()
    mon.notifyAll()
    mon.unlock()

    /* Wait for the new thread to complete */
    fut.get()
    return 0
}

运行结果：

New thread: before wait
Main thread: set flag
Main thread: notify
New thread: after wait

从用法上看，Monitor 与"Mutex + condition()"非常相似：wait() 挂起并释放锁，notifyAll() 唤醒所有等待线程。Monitor.wait() 还支持命名参数 timeout!: Duration，默认 Duration.Max（无限等待），返回 Bool 表示是被信号唤醒还是超时。

4.2 为什么废弃？迁移到 Condition

查看 stdlib/libs/std/sync/monitor.cj 的源码，Monitor 类声明为：

@Deprecated[message: "Use `public interface Condition` instead."]
public class Monitor <: ReentrantMutex {
    ...
}

源码与 sync_package_classes.md 均提示：请改用 Condition 接口。原因是职责分离的设计演进：Condition 把"锁"（Mutex）与"等待队列"（Condition）解耦，同一把 Mutex 可以通过 condition() 生成多个条件变量，实现"单锁多等待队列"的并发原语；而 Monitor 把两者耦合在一起，灵活性受限。同被废弃的还有 MultiConditionMonitor、ReentrantMutex、ReentrantReadWriteMutex 等旧类，均以 @Deprecated 标注并给出替代类型。

因此，新代码应直接使用 Mutex + Condition（见下一节），仅维护旧代码时才需要理解 Monitor。

五、Condition 条件变量：推荐的线程挂起/唤醒方案

5.1 完整示例：基于谓词的 waitUntil

Condition 接口提供 wait()、wait(timeout!)、waitUntil(predicate)、waitUntil(predicate, timeout!)、notify()、notifyAll() 六个方法。相比 Monitor 的 while (flag) { wait() } 手写循环，waitUntil 把"检查条件 → 挂起 → 被唤醒后再检查"的循环封装进 API，代码更简洁、更不易出错：

import std.sync.{Mutex, Condition, AtomicBool}

var mutex = Mutex()
var flag = AtomicBool(true)

main(): Int64 {
    let condition: Condition
    synchronized(mutex) {
        condition = mutex.condition() // Generate Condition instance while holding the lock
    }

    let fut = spawn {
        synchronized(mutex) {
            println("New thread: before wait")
            condition.waitUntil {=> !flag.load()} // Suspend current thread until flag becomes true
            println("New thread: after wait")
        }
    }

    /* Sleep for 10 milliseconds to ensure the new thread can execute */
    sleep(10 * Duration.millisecond)

    synchronized(mutex) { // Wait for child thread to suspend
        println("Main thread: set flag")
        flag.store(false) // Modify flag value
        println("Main thread: notify")
        condition.notifyAll() // Wake up waiting child thread
    }

    /* Wait for the new thread to complete */
    fut.get()
    return 0
}

运行结果：

New thread: before wait
Main thread: set flag
Main thread: notify
New thread: after wait

要点解读：

• condition = mutex.condition() 同样要求在持锁状态下调用，示例用 synchronized(mutex) 保证这一点；
• condition.waitUntil {=> !flag.load()} 的语义（见 sync_package_interfaces.md）：先检查谓词，若为 true 立即返回；否则挂起当前线程并释放锁，直到被 notifyAll() 唤醒且谓词求值为 true 才返回。这天然避免了"唤醒后条件仍未满足"的虚假唤醒问题；
• 主线程在持锁状态下修改共享状态（flag.store(false)）并调用 notifyAll()，符合"修改共享变量后再唤醒等待者"的规范用法。

5.2 源码剖析：MutexCondition 与 Condition 接口

stdlib/libs/std/sync/lock_interface.cj 定义了完整的 Condition 接口，其注释点明它是"与互斥锁关联的等待队列（wait queue）"。而 Mutex.condition() 返回的实际实现类 MutexCondition 定义在 mutex_cjnative.cj 中：

class MutexCondition <: Condition {
    private let condition: Monitor

    init(mutex: Mutex) {
        condition = Monitor(mutex)
    }

    public func waitUntil(predicate: () -> Bool): Unit {
        condition.checkMutexStatus()
        while (!predicate()) {
            monitorWait(condition, MAX_TIMEOUT_NS)
        }
    }
    ...
}

从源码可以看到 waitUntil 的实现本质是一个 while (!predicate()) monitorWait(...) 循环，这与 Monitor 示例中手写的 while (flag) { mon.wait() } 完全同构——waitUntil 只是把这一惯用法封装成了接口方法。wait()、notify()、notifyAll() 最终都委托给 monitorWait/monitorNotify/monitorNotifyAll 这些在 native.cj 中声明的 @Intrinsic 函数，并由运行时在 Sync.cpp 中落地实现（MCC_MonitorWait 等）。

另外，带超时的 waitUntil(predicate, timeout!) 实现会记录起始时间（MonoTime.now()），在循环中扣减已消耗时间，超时则返回 false；timeout 若小于等于 Duration.Zero 会抛出 IllegalArgumentException。

六、Timer 定时器：一次性与重复性任务

6.1 完整示例：once + repeat 的组合

Timer 用于在指定时间点或间隔后执行任务，分为一次性定时器与重复定时器。示例同时创建了一个 50ms 后执行一次的任务，以及一个延迟 100ms、每 200ms 重复执行的任务，并通过 cancel() 停止重复任务：

import std.sync.*

main(): Int64 {
    let count = AtomicInt8(0)

    Timer.once(50 * Duration.millisecond) {
        =>
        println("run only once")
        count.fetchAdd(1)
    }

    let timer = Timer.repeat(
        100 * Duration.millisecond,
        200 * Duration.millisecond,
        {
            =>
            println("run repetitively")
            count.fetchAdd(10)
        }
    )

    sleep(Duration.second)
    timer.cancel()
    sleep(500 * Duration.millisecond)
    println("count = ${count.load()}")
    0
}

运行结果：

run only once
run repetitively
run repetitively
run repetitively
run repetitively
run repetitively
count = 51

结果解读与验证：

• Timer.once(50ms, ...) 在 50ms 后执行一次，count += 1；
• Timer.repeat(100ms, 200ms, ...) 在 100ms、300ms、500ms、700ms、900ms 共执行 5 次（主线程 sleep(Duration.second) 的 1 秒窗口内），count += 10 共 5 次；
• 因此 count = 1 + 5 × 10 = 51，与输出完全吻合。cancel() 后重复任务不再执行，sleep(500ms) 用于验证取消确实生效。

6.2 源码剖析：Timer 的调度机制与 CatchupStyle

stdlib/libs/std/sync/timer.cj 中 Timer 类的注释揭示了其关键语义：

• 每个 Timer 内部隐含一次 spawn，即每个 Timer 创建一个新线程执行其关联任务；
• 一个 Timer 初始化时只能绑定一个任务，绑定后不可重置；
• Timer 的生命周期：只有当关联任务执行完成或被 cancel() 主动取消后，Timer 实例才结束生命周期并可被 GC 回收；在此之前 GC 不会回收它，从而保证任务必然被执行；
• 系统繁忙时不保证任务准时触发：Timer 只保证"触发时间不超过当前时间"，后续任务会相应顺延；
• Timer 不捕获任务抛出的异常，任务一旦抛出未捕获异常，该 Timer 即失效。

Timer 提供的静态工厂方法（对应 sync_package_classes.md 的 API 文档）包括：

方法	语义
Timer.after(delay, task)	任务返回 Option<Duration>：返回 None 则停止调度；返回 Some(v) 且 v > Duration.Zero 则以 v 为下次最小间隔，否则立即再次调度
Timer.once(delay, task)	一次性任务，只执行一次
Timer.repeat(delay, interval, task, style!)	周期性任务，默认 style: Burst；interval <= Duration.Zero 时抛 IllegalArgumentException
Timer.repeatDuring(period, delay, interval, task, style!)	在 period 时间窗口内周期性执行
Timer.repeatTimes(count, delay, interval, task, style!)	周期执行固定次数；count <= 0 时抛 IllegalArgumentException

其中 Timer.repeat 的第一个参数是首次执行的延迟（delay），第二个参数是两次执行之间的最小间隔（interval），示例中 100ms/200ms 分别对应这两者。

周期性定时器的追赶策略由 CatchupStyle 枚举 控制，共三种：

• Burst（默认）：任务开始时间点之间的间隔固定。当任务执行时间超过调度间隔时，错过的任务会被依次"补执行"直到追赶上进度。适合"任务之间存在顺序依赖、必须按序执行"的场景。
• Delay：上一个任务结束到下一个任务开始之间的间隔固定，即"下次开始时间 = 上次结束时间 + 间隔"。适合"不关心任务精确开始时间、但要求任务间间隔固定"的场景。
• Skip：任务开始时间点之间的间隔固定，但当任务执行时间超过间隔时，跳过后续错过的任务以尽快追赶。适合"任务相互独立、可以跳过中间任务"的场景。

从 timer.cj 的实现看，三种策略分别由 burstStyleTick、delayStyleTick、skipStyleTick 三个函数生成"下一次触发间隔"的计算逻辑，Timer.repeat 通过 match (style) 选择对应策略；repeatTimes/repeatDuring 则在此基础上叠加次数/时间窗口判断。

调度器方面，源码中 TimerScheduler 共创建 64 个调度器实例，按 Timer 实例 ID 取模分发（getScheduler(id)），每个调度器内部包含一个 Monitor、一个最小堆 TimerHeap 与一个无锁队列 TimerQueue；TimerScheduler.start() 通过 spawn 启动一个循环线程：从队列消费任务加入堆、取堆顶到期任务执行、未到期则 monitor.wait(timeout: endTime - now) 精确等待。取消的 Timer 不会立即从堆中移除，而是累计 cancelCount，当"取消数 > 堆大小 / 4 且 > 10"时执行一次 clearCancelledTimers() 清理（源码注释说明这是业界通用做法）。可见 Timer 的整体调度建立在 spawn 线程与 Monitor 条件等待之上。

6.3 使用注意

• cancel() 不会中断正在执行的任务：如果调用 cancel 时任务正在运行，当前这次运行不会被中断，且 cancel() 不阻塞当前线程；重复调用 cancel 与调用一次等价（见 timer.cj 的 cancel 实现，通过 isEnabled 标志保证幂等）。
• 若任务可能抛出异常，应在任务内部自行 try/catch，否则整个 Timer 将失效。
• 任务触发时间受系统负载影响，不适合作为精确计时的依据。

七、参考资料与进一步阅读

本文对应的官方示例文档与相关 API 参考均位于当前仓库，可继续深入：

• 示例文档（本文主线）：stdlib/doc/libs/std_en/sync/sync_samples/sync_samples.md（中文版见 stdlib/doc/libs/std/sync/sync_samples/sync_samples.md）
• 包概览与 synchronized 说明：stdlib/doc/libs/std_en/sync/sync_package_overview.md
• 类参考（原子类型、Mutex、Timer 等）：stdlib/doc/libs/std_en/sync/sync_package_api/sync_package_classes.md
• 接口参考（Lock、Condition、UniqueLock）：stdlib/doc/libs/std_en/sync/sync_package_api/sync_package_interfaces.md
• 枚举参考（CatchupStyle、MemoryOrder）：stdlib/doc/libs/std_en/sync/sync_package_api/sync_package_enums.md
• 标准库源码：stdlib/libs/std/sync（原子操作 atomic.cj、互斥锁 mutex_cjnative.cj、条件变量 lock_interface.cj、定时器 timer.cj、运行时内置函数声明 native.cj）
• 运行时同步原语实现：runtime/src/Sync/Sync.cpp 与 runtime/src/Sync/Sync.h

实践建议：新代码优先使用 Atomic* 原子类型 + Mutex/synchronized + Condition（尤其 waitUntil） 的组合；遇到 Monitor、ReentrantMutex、MemoryOrder 等标注 deprecated 的 API 时，按文档指引迁移到对应的现代替代接口。