1 概述
1.1 案例介绍
仓颉编程语言作为一款面向全场景应用开发的现代编程语言，通过现代语言特性的集成、全方位的编译优化和运行时实现、以及开箱即用的 IDE 工具链支持，为开发者打造友好开发体验和卓越程序性能。
案例结合代码体验，帮助大家更直观的学习仓颉语言中并发编程知识。
1.2 适用对象
1.3 案例时间
本案例总时长预计 60 分钟。
1.4 案例流程
说明：
进入华为开发者空间，登录云主机；
使用 CodeArts IDE for Cangjie 编程和运行仓颉代码。。
2 环境准备
2.1 开发者空间配置
面向广大开发者群体，华为开发者空间提供一个随时访问的“开发桌面云主机”、丰富的“预配置工具集合”和灵活使用的“场景化资源池”，开发者开箱即用，快速体验华为根技术和资源。
领取云主机后可以直接进入华为开发者空间工作台界面，点击打开云主机 > 进入桌面连接云主机。没有领取在开发者空间根据指引领取配置云主机即可，云主机配置参考 1.5 资源总览。
点击桌面 CodeArts IDE for Cangjie，打开编辑器，点击新建工程，名称 demo，其他保持默认配置，点击创建。
产物类型说明：
创建完成后，打开 src/main.cj，点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
后续文档中的代码验证均可以替换 main.cj 中的代码（package demo 包路径保留）后执行，demo 是项目名称，与创建项目时设置的保持一致。
至此，云主机环境配置完毕。
3 并发编程
3.1 并发概述
并发编程是现代编程语言中不可或缺的特性，仓颉编程语言提供抢占式的线程模型作为并发编程机制。在谈及编程语言和线程时，线程可以细化为两种不同概念：语言线程 和 native 线程。
语言线程：是编程语言中并发模型的基本执行单位，语言线程的目的是屏蔽底层实现细节。
native 线程：指语言实现中所使用到的线程（一般是操作系统线程），他们作为语言线程的具体实现载体。
仓颉线程本质上是用户态的轻量级线程，每个仓颉线程都受到底层 native 线程的调度执行，并且多个仓颉线程可以由一个 native 线程执行。每个 native 线程会不断地选择一个就绪的仓颉线程完成执行，如果仓颉线程在执行过程中发生阻塞（例如等待互斥锁的释放），那么 native 线程会将当前的仓颉线程挂起，并继续选择下一个就绪的仓颉线程。发生阻塞的仓颉线程在重新就绪后会继续被 native 线程调度执行。
在大多数情况下，开发者只需要面向仓颉线程进行并发编程而不需要考虑这些细节。但在进行跨语言编程时，开发者需要谨慎调用可能发生阻塞的 foreign 函数，例如 IO 相关的操作系统调用等。
例如：下列示例代码中的新线程会调用 foreign 函数 socket_read。在程序运行过程中，某一 native 线程将调度并执行该仓颉线程，在进入到 foreign 函数中后，系统调用会直接阻塞当前 native 线程直到函数执行完成。native 线程在阻塞期间将无法调度其他仓颉线程来执行，这会降低程序执行的吞吐量。
foreign socket_read(sock: Int64): CPointer<Int8>let fut = spawn { let sock: Int64 = ... let ptr = socket_read(sock)}
复制代码
3.2 创建线程
当开发者希望并发执行某一段代码时，只需创建一个仓颉线程即可。要创建一个新的仓颉线程，可以使用关键字 spawn 并传递一个无形参的 lambda 表达式，该 lambda 表达式即为在新线程中执行的代码。
例如在主线程中新建一个线程，两线程分别打印文本。
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*main(): Int64 { // 创建新线程 spawn { => println("New thread before sleeping") // 新线程打印文本 sleep(100 * Duration.millisecond) // 新线程睡眠100ms. println("New thread after sleeping") } // 主线程打印文本 println("Main thread") return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
（* 注意：在上面的例子中，新线程会在主线程结束时一起停止，无论这个新线程是否已完成运行，所以每次运行的结果可能不一样）
sleep()函数：可以让当前线程睡眠指定的时间后再恢复执行，其时间由指定的 Duration 类型决定。函数原型为：
func sleep(dur: Duration): Unit
复制代码
（* 注意：Duration.Zero 表示 0 纳秒时间间隔的 Duration 实例，如果 dur <= Duration.Zero, 那么当前线程只会让出执行资源，并不会进入睡眠）
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*main(): Int64 { println("Hello") // Duration.second表示1秒时间间隔的Duration实例 sleep(3 * Duration.second) // 让主线程睡眠 3s. println("World") return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.3 访问线程
3.3.1 使用 Future<T> 等待线程结束
在上面的例子中，新创建的线程会由于主线程结束而提前结束，在缺乏顺序保证的情况下，甚至可能会出现新创建的线程还来不及得到执行就退出了。可以通过 spawn 表达式的返回值，来等待线程执行结束。
spawn 表达式的返回类型是 Future<T>，其中 T 是类型变元，其类型与 lambda 表达式的返回类型一致。当调用 Future<T> 的 get() 成员函数时，它将等待它的线程执行完成。
下方示例代码演示了如何使用 Future<T> 在 main 中等待新创建的线程执行完成：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*main(): Int64 { let fut: Future<Unit> = spawn { => println("New thread before sleeping") sleep(3 * Duration.second) // 睡眠3s. println("New thread after sleeping") } fut.get() // 等待该线程执行完成. println("Main thread") return 0}
复制代码
get()调用后的代码会等待调用线程执行完成后再执行。
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.3.2 使用 Future<T> 获取线程返回值
Future<T> 除了可以用于阻塞等待线程执行结束以外，还可以获取线程执行的结果。下面我们来看一下它提供的具体成员函数：
1. get(): T：阻塞等待线程执行结束，并返回执行结果，如果该线程已经结束，则直接返回执行结果，示例如下：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*main(): Int64 { let fut: Future<Int64> = spawn { sleep(Duration.second) // 睡眠 1s. return 1 } try { // 等待线程fut执行完成并获取执行结果 let res: Int64 = fut.get() println("result = ${res}") } catch (_) { println("线程执行出现异常") } return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
2. **get(ns: Int64): Option<T>**：阻塞等待该 Future<T> 所代表的线程执行结束，并返回执行结果，当到达超时时间 ns 时，如果该线程还没有执行结束，将会返回 Option<T>.None。如果 ns <= 0,其行为与 get() 相同，示例如下：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*main(): Int64 { let fut = spawn { sleep(Duration.second) // 睡眠 1s. return 1 } // 等待fut线程执行完成并获取结果, 等待超时时间 1ms. let res: Option<Int64> = fut.get(1000 * 1000) match (res) { case Some(val) => println("result = ${val}") case None => println("等待fut执行完成超时") } return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.3.3 访问线程属性
每个 Future<T> 对象都有一个对应的仓颉线程，以 Thread 对象为表示。Thread 类主要被用于访问线程的属性信息，例如线程标识等。需要注意的是，Thread 无法直接被实例化构造对象，仅能从 Future<T> 的 thread 成员属性获取对应的 Thread 对象，或是通过 Thread 的静态成员属性 currentThread 得到当前正在执行线程对应的 Thread 对象。
例如，在创建新线程后分别通过两种方式获取线程标识。由于主线程和新线程获取的是同一个 Thread 对象，所以他们能够打印出相同的线程标识。
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*main(): Unit { let fut = spawn { println("Current thread id: ${Thread.currentThread.id}") } println("New thread id: ${fut.thread.id}") fut.get()}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.4 终止线程
可以通过 Future<T> 的 cancel() 方法向对应的线程发送终止请求，该方法不会停止线程执行。开发者需要使用 Thread 的 hasPendingCancellation 属性来检查线程是否存在终止请求。
一般而言，如果线程存在终止请求，那么开发者可以实施相应的线程终止逻辑。因此，如何终止线程都交由开发者自行处理，如果开发者忽略终止请求，那么线程继续执行直到正常结束。
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.SyncCountermain(): Unit { let syncCounter = SyncCounter(1) let fut = spawn { syncCounter.waitUntilZero() // 等待倒数计数器变为0 // 检查取消请求, 自定义取消逻辑 if (Thread.currentThread.hasPendingCancellation) { println("cancelled") return } println("hello") } fut.cancel() // 发送取消请求 syncCounter.dec() // 唤醒所有等待的线程 fut.get() // 确保fut线程执行完成}
复制代码
类型 SyncCounter 提供倒数计数器功能，线程可以等待计数器变为零。
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.5 同步机制
在并发编程中，如果缺少同步机制来保护多个线程共享的变量，很容易会出现数据竞争问题（data race）。
仓颉编程语言提供三种常见的同步机制来确保数据的线程安全：原子操作、互斥锁和条件变量。
3.5.1 原子操作 Atomic
仓颉提供整数类型、Bool 类型和引用类型的原子操作。
其中整数类型包括： Int8、Int16、Int32、Int64、UInt8、UInt16、UInt32、UInt64。整数类型的原子操作支持基本的读写、交换以及算术运算操作：
（* 注意：交换操作和算术操作的返回值是修改前的值；compareAndSwap 是判断当前原子变量的值是否等于 old 值，如果等于，则使用 new 值替换；否则不替换）
例如，在多线程程序中，使用原子操作实现计数：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*import std.collection.*let count = AtomicInt64(0)main(): Int64 { let list = ArrayList<Future<Int64>>() // 创建1000个线程 for (_ in 0..1000) { let fut = spawn { sleep(Duration.millisecond) // 睡眠 1ms. count.fetchAdd(1) } list.append(fut) } // 等待所有线程执行完成 for (f in list) { f.get() } let val = count.load() println("count = ${val}") return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
Bool 类型和引用类型的原子操作只提供读写和交换操作：
原子引用类型是 AtomicReference，以下是使用 Bool 类型、引用类型原子操作的一些正确示例：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*class A {}main() { var obj = AtomicBool(true) var x1 = obj.load() // x1: true, 类型是 Bool println(x1) var t1 = A() var obj2 = AtomicReference(t1) var x2 = obj2.load() // x2 和 t1 是相同的对象 // 相同的对象, 交换成功, y1: true var y1 = obj2.compareAndSwap(x2, t1) println(y1) var t2 = A() // 不是相同的对象, 交换失败, y2: false var y2 = obj2.compareAndSwap(t2, A()) println(y2) y2 = obj2.compareAndSwap(t1, A()) // 交换成功, y2: true println(y2)}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.5.2 可重入互斥锁 ReentrantMutex
可重入互斥锁的作用是对临界区加以保护，使得任意时刻最多只有一个线程能够执行临界区的代码。当一个线程试图获取一个已被其他线程持有的锁时，该线程会被阻塞，直到锁被释放，该线程才会被唤醒，可重入是指线程获取该锁后可再次获得该锁。
ReentrantMutex 是仓颉内置的互斥锁。使用可重入互斥锁时，必须牢记两条规则：
ReentrantMutex 提供的主要成员函数如下：
public open class ReentrantMutex { // 创建可重入互斥锁 public init() // 锁定互斥体，如果互斥体已被锁定，则阻塞 public func lock(): Unit // 解锁互斥体 // 如果有其他线程阻塞在此锁上，那么唤醒他们中的一个 public func unlock(): Unit // 尝试锁定互斥体 // 如果互斥体已被锁定，则返回 false；反之，则锁定互斥体并返回 true public func tryLock(): Bool}
复制代码
例如，使用 ReentrantMutex 来保护对全局共享变量 count 的访问，对 count 的操作即属于临界区：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*import std.collection.*var count: Int64 = 0let mtx = ReentrantMutex()main(): Int64 { let list = ArrayList<Future<Unit>>() // 创建100个线程. for (_ in 0..1000) { let fut = spawn { sleep(Duration.millisecond) // 睡眠 1ms. mtx.lock() // 上锁 count++ // 临界区代码 mtx.unlock() // 释放锁 } list.append(fut) } // 等待所有线程完成 for (f in list) { f.get() } println("count = ${count}") return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.5.3 Monitor
Monitor 是一个内置的数据结构，它绑定了互斥锁和单个与之相关的条件变量（也就是等待队列）。Monitor 可以使线程阻塞并等待来自另一个线程的信号以恢复执行。这是一种利用共享变量进行线程同步的机制，主要提供如下方法：
public class Monitor <: ReentrantMutex { // 通过默认构造函数创建 Monitor public init() // 当前线程挂起，直到对应的 notify 函数被调用，或者挂起时间超过 timeout public func wait(timeout!: Duration = Duration.Max): Bool // 唤醒一个等待在该 Montior 上的线程 public func notify(): Unit // 唤醒所有等待在该 Montior 上的线程 public func notifyAll(): Unit}
复制代码
下面是一个使用 Monitor 实现互斥锁的示例
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.time.*var mon = Monitor()var flag: Bool = truemain(): Int64 { let fut = spawn { mon.lock() // 上锁 while (flag) { println("New thread: before wait") mon.wait() println("New thread: after wait") } mon.unlock() // 解锁 } // 睡眠 10ms, 确保新线程执行完成. sleep(10 * Duration.millisecond) mon.lock() println("Main thread: set flag") flag = false mon.unlock() mon.lock() println("Main thread: notify") mon.notifyAll() mon.unlock() // 等待新线程执行完成 fut.get() return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.5.4 MultiConditionMonitor
MultiConditionMonitor 是一个内置的数据结构，它绑定了互斥锁和一组与之相关的动态创建的条件变量。该类应仅当在 Monitor 类不足以满足复杂的线程间同步的场景下使用。主要提供如下方法：
public class MultiConditionMonitor <: ReentrantMutex { // 通过默认构造函数创建 MultiConditionMonitor init() // 创建一个与该 Monitor 相关的 ConditionID // 可能被用来实现 “单互斥体多等待队列” 的并发原语 // 如果当前线程没有持有该互斥体，抛出异常 func newCondition(): ConditionID // 当前线程挂起，直到对应的 notify 函数被调用 func wait(id: ConditionID, timeout!: Duration = Duration.Max): Bool // 唤醒等待在所指定的条件变量的线程（如果有） func notify(id: ConditionID): Unit // 唤醒所有等待在所指定的条件变量的线程（如果有） func notifyAll(id: ConditionID): Unit}
复制代码
（* 初始化时，MultiConditionMonitor 没有与之相关的 ConditionID 实例。每次调用 newCondition 都会将创建一个新的条件变量并与当前对象关联）
例如，使用 MultiConditionMonitor 去实现一个长度固定的有界 FIFO(先进先出) 队列，当队列为空，get() 会被阻塞；当队列满了时，put() 会被阻塞。
import std.sync.*class BoundedQueue { // 创建一个 MultiConditionMonitor, 两个Conditions. let m: MultiConditionMonitor = MultiConditionMonitor() var notFull: ConditionID var notEmpty: ConditionID var count: Int64 // 整数缓冲区 var head: Int64 // 写入索引 var tail: Int64 // 读取索引 // 队列长度100 let items: Array<Object> = Array<Object>(100, {i => Object()}) init() { count = 0 head = 0 tail = 0 synchronized(m) { notFull = m.newCondition() notEmpty = m.newCondition() } } // 插入一个对象，如果队列已满，则使当前线程阻塞。 public func put(x: Object) { // 加互斥锁 synchronized(m) { while (count == 100) { // 如果队列已满, 等待 "queue notFull" 事件触发 m.wait(notFull) } items[head] = x head++ if (head == 100) { head = 0 } count++ // 已经插入了一个对象，并且当前队列不再是空的， // 因此唤醒之前由于队列是空的而被get()阻塞的线程 m.notify(notEmpty) } // 释放互斥锁 } // 如果队列为空，则弹出一个对象，并使当前线程阻塞 public func get(): Object { // 加互斥锁 synchronized(m) { while (count == 0) { // 如果队列为空, 等待 "queue notEmpty" 事件触发 m.wait(notEmpty) } let x: Object = items[tail] tail++ if (tail == 100) { tail = 0 } count-- // 弹出一个对象，而当前队列不再满， // 因此唤醒之前由于队列已满而被put()阻塞的线程 m.notify(notFull) return x } // 释放互斥锁 }}
复制代码
3.5.5 synchronized 关键字
互斥锁 ReentrantMutex 提供了一种便利灵活的加锁的方式，同时因为它的灵活性，也可能引起忘了解锁，或者在持有互斥锁的情况下抛出异常不能自动释放持有的锁的问题。因此，仓颉编程语言提供一个 synchronized 关键字，搭配 ReentrantMutex 一起使用，可以在其后跟随的作用域内自动进行加锁解锁操作，用来解决类似的问题。
注意：一个线程在进入 synchronized 修饰的代码块之前，会自动获取 ReentrantMutex 实例对应的锁，如果无法获取锁，则当前线程被阻塞。而线程在退出 synchronized 修饰的代码块之前，会自动释放该 ReentrantMutex 实例的锁，如通过控制转移表达式（如 break、continue、return、throw）跳出 synchronized 代码块。
使用 synchronized 关键字来保护共享数据可参考：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
import std.sync.*import std.collection.*var count: Int64 = 0var mtx: ReentrantMutex = ReentrantMutex()main(): Int64 { let list = ArrayList<Future<Unit>>() for (_ in 0..10) { let fut = spawn { while (true) { // 使用 synchronized(mtx), 替换mtx.lock() 和 mtx.unlock(). synchronized(mtx) { count = count + 1 break // 由于break跳出while循环，包括synchronized 代码块， // 所以新线程中以下打印语句不会执行 println("in thread") } } } list.append(fut) } // 等待所有线程执行完成 for (f in list) { f.get() } synchronized(mtx) { println("in main, count = ${count}") } return 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
3.5.6 线程局部变量 ThreadLocal
使用 core 包中的 ThreadLocal 可以创建并使用线程局部变量，使用 ThreadLocal 实际是一种以空间换时间的做法，每一个线程都有它独立的一个存储空间来保存这些线程局部变量，因此，在每个线程可以安全地访问他们各自的线程局部变量，而不需要等待其他线程释放锁。主要提供如下方法：
public class ThreadLocal<T> { // 构造一个携带空值的仓颉线程局部变量 public init() // 获得仓颉线程局部变量的值，如果值不存在，则返回 Option<T>.None public func get(): Option<T> // 通过 value 设置仓颉线程局部变量的值 public func set(value: Option<T>): Unit}
复制代码
例如，两个线程通过 ThreadLocal 类来创建并使用各自线程的局部变量：
Step1：复制以下代码，替换 main.cj 文件中的代码。（保留 package）
main(): Int64 { let tl = ThreadLocal<Int64>() let fut1 = spawn { tl.set(123) println("tl in spawn1 = ${tl.get().getOrThrow()}") } let fut2 = spawn { tl.set(456) println("tl in spawn2 = ${tl.get().getOrThrow()}") } fut1.get() fut2.get() 0}
复制代码
Step2：点击编辑器右上角运行按钮直接运行，终端窗口可以看到打印内容。
至此，仓颉语言中并发编程知识内容介绍告一段落。
如果想了解更多仓颉编程语言知识可以访问： https://cangjie-lang.cn/
评论