合理使用多线程共享内存
概述
在应用开发中，为了避免主线程阻塞，提高应用性能，需要将一些耗时操作放在子线程中执行。此时，子线程就需要访问主线程中的数据。ArkTS采用了基于消息通信的Actor并发模型，具有内存隔离的特性，所以跨线程传输数据时需要将数据序列化，但是AkrTS支持通过可共享对象SharedArrayBuffer实现直接的共享内存。
在开发应用时，如果遇到数据量较大，并且需要多个线程同时操作的情况，推荐使用SharedArrayBuffer共享内存，可以减少数据在线程间传递时需要复制和序列化的额外开销。比如，音视频解码播放、多个线程同时读取写入文件等场景。由于内存是共享的，所以在多个线程同时操作同一块内存时，可能会引起数据的紊乱，这时就需要使用锁来确保数据操作的有序性。本文将基于此具体展开说明。关于多线程的使用和原理，可参考OpenHarmony多线程能力场景化示例实践，本文将不再详细讲述。
工作原理
可共享对象SharedArrayBuffer，是拥有固定长度的原始二进制数据缓冲区，可以存储任何类型的数据，包括数字、字符串等。它支持在多线程之间传递，传递之后的SharedArrayBuffer对象和原始的SharedArrayBuffer对象可以指向同一块内存，进而达到共享内存的目的。SharedArrayBuffer对象存储的数据在子线程中被修改时，需要通过原子操作保证其同步性，即下个操作开始之前务必需要保证上个操作已经结束。下面将通过示例说明原子操作保证同步性的必要性，详细代码请参考AtomicsUsage.ets。
非原子操作
......
// 非原子操作，进行10000次++
@Concurrent
function normalProcess(int32Array: Int32Array) {
for (let i = 0; i < 10000; i++) {
int32Array[0]++;
}
}
// 原子操作，进行10000次++
@Concurrent
function atomicsProcess(int32Array: Int32Array) {
for (let i = 0; i < 10000; i++) {
Atomics.add(int32Array, 0, 1);
}
}
......
@State result: string = "计算结果：";
private taskNum: number = 2;
private scroller: Scroller = new Scroller();
......
Button("非原子操作")
.width("80%")
.fontSize(30)
.fontWeight(FontWeight.Bold)
.margin({ top: 30 })
.onClick(async () => {
this.sharedArrayBufferUsage(false);
})
Scroll(this.scroller) {
Column() {
Text(this.result)
.width("80%")
.fontSize(30)
.fontWeight(FontWeight.Bold)
.fontColor(Color.Blue)
}
}.height("60%")
.margin({ top: 30 })
......
// 根据传入的值isAtomics判断是否使用原子操作
sharedArrayBufferUsage(isAtomics: boolean) {
// 创建长度为4的SharedArrayBuffer对象
let sab: SharedArrayBuffer = new SharedArrayBuffer(4);
// 由于SharedArrayBuffer是原始二进制数据缓冲区，无法直接使用，所以这里转换为Int32Array类型进行后续操作
let int32Array: Int32Array = new Int32Array(sab);
int32Array[0] = 0;
let taskGroup: taskpool.TaskGroup = new taskpool.TaskGroup();
// 创建Task对象，并放入TaskGroup中执行
for (let i = 0; i < this.taskNum; i++) {
if (isAtomics) {
taskGroup.addTask(new taskpool.Task(atomicsProcess, int32Array));
} else {
taskGroup.addTask(new taskpool.Task(normalProcess, int32Array));
}
}
taskpool.execute(taskGroup).then(() => {
// 将结果打印在Text上
this.result = this.result + "\n" + int32Array;
// 如果Scroll不在最低端，则滑动到最低端
if (!this.scroller.isAtEnd()) {
this.scroller.scrollEdge(Edge.Bottom);
}
}).catch((e: BusinessError) => {
logger.error(e.message);
})
}
在这段代码中，创建了2个task，对SharedArrayBuffer分别进行了10000次自增操作，预期的结果应该是20000。点击按钮查看计算结果，就会发现最后的结果并不一定是20000，并且每次点击后，计算的结果都可能是不同的。 这是因为SharedArrayBuffer是共享内存的，多个线程同时进行自增时，是操作的同一块内存，而自增操作并不是原子操作，需要经过以下三个步骤：
- 第一步，从内存中取值
- 第二步，对取出的值+1
- 第三步，将结果写入内存
当多个线程同时操作时，就会发生这样一种情况：A线程在第一步取值1000，第二步+1操作后是1001，在执行第三步之前，B线程也去取值了，这时由于A线程还没有将结果写入内存，所以B线程取到的值依然是1000，然后A执行第三步将1001写入了内存，而B会对1000进行+1操作并将结果1001写入同一块内存。这样就会导致明明进行了两次+1的操作，但是结果并没有变成预期的1002，而是1001。所以在这个示例中会出现结果不符合预期的情况。
原子操作
下面修改一下代码，将自增操作改为使用Atomics.add()方法的原子操作。
......
Button("原子操作")
.width("80%")
.fontSize(30)
.fontWeight(FontWeight.Bold)
.margin({ top: 30 })
.onClick(async () => {
this.sharedArrayBufferUsage(true);
})
......
点击按钮查看计算结果，就会发现不论计算多少次，结果一直都是20000。这是因为，原子操作是不可中断的一个或者一系列操作，可以保证在A线程执行完取值、计算、写入内存这三个步骤之前，不会被B线程中断，也就不会发生非原子操作示例中B线程取到旧值的情况，而是每次都能拿到A线程写入内存的新值。所以，在使用SharedArrayBuffer共享内存时，一定要注意使用原子操作保证同步性，否则就可能会造成数据的紊乱。
场景示例
在应用开发中使用多线程时，会遇到处理复杂逻辑的情况，是无法保证整个线程都是一个原子操作的，此时就可以使用锁来解决一段代码的原子性问题。
锁的实现
并发编程重在解决线程间分工、同步与互斥的问题，而实现互斥的重要方式是通过锁。示例通过Atomics和SharedArrayBuffer简单实现不可重入锁类NonReentrantLock。 constructor()通过传入可共享对象SharedArrayBuffer初始化锁，实现多线程共享同一块内存，以作为共同操作的标志位，从而控制锁的状态。
const UNLOCKED = 0;
const LOCKED_SINGLE = 1;
const LOCKED_MULTI = 2;
export class NonReentrantLock {
flag: Int32Array;
constructor(sab: SharedArrayBuffer) { // 传入一个4bytes的SharedArrayBuffer
this.flag= new Int32Array(sab); // 其视图为只有一位的int数组（1 = 4bytes * 8 / 32bit）
}
lock(): void {...}
tryLock(): boolean {...}
unlock(): void {...}
}
lock()方法用于获取锁，如果获取锁失败，则线程进入阻塞状态。
lock(): void {
const flag= this.flag;
let c = UNLOCKED;
// 如果flag数组的0位置，当前值为UNLOCKED，则改为LOCKED_SINGLE；否则，进入do-while循环，阻塞线程
if ((c = Atomics.compareExchange(flag, 0, UNLOCKED, LOCKED_SINGLE)) !== UNLOCKED) {
do {
// 有线程拿不到锁时，修改标志位为LOCKED_MULTI，并使之进入睡眠阻塞状态
if (c === LOCKED_MULTI || Atomics.compareExchange(flag, 0, LOCKED_SINGLE, LOCKED_MULTI) !== UNLOCKED) {
Atomics.wait(flag, 0, LOCKED_MULTI);
}
// 被唤醒的线程，如果还是没有拿到锁，就回到循环中，重新进入阻塞状态
} while ((c = Atomics.compareExchange(flag, 0, UNLOCKED, LOCKED_MULTI)) !== UNLOCKED);
}
}
tryLock()方法用于尝试获取锁，如果获取锁成功则返回true，失败返回false，但不会阻塞线程。
tryLock(): boolean {
const flag= this.flag;
return Atomics.compareExchange(flag, 0, UNLOCKED, LOCKED_SINGLE) === UNLOCKED;
}
unlock()方法用于释放锁。
unlock(): void {
// 局部化flag，保证只有获取锁的线程可以释放锁
const flag= this.flag;
let v0 = Atomics.sub(flag, 0, 1);
if (v0 !== LOCKED_SINGLE) {
Atomics.store(flag, 0, UNLOCKED);
// 只唤醒在数组0索引位置等待的其中一个线程，去上方lock()方法while条件中检测，尝试获取锁
Atomics.notify(flag, 0, 1);
}
}
锁的应用
示例通过多线程写入文件的场景，展示多线程不合理操作共享内存时，出现的线程不安全问题，进而导致输出文件乱码的情况。并通过使用上文实现的NonReentrantLock，解决该问题。 主线程通过startWrite(useLock: boolean)方法，开启多线程写入文件，并通过useLock参数控制是否使用锁。
@Component
export struct LockUsage {
taskNum: number = 10; // 任务数，实际并行线程数依设备而定
baseDir: string = getContext().filesDir + '/TextDir'; // 文件写入的应用沙箱路径
sabInLock: SharedArrayBuffer = new SharedArrayBuffer(4); // 在主线程，初始化子线程锁标志位，所使用的共享内存
sabForLine: SharedArrayBuffer = new SharedArrayBuffer(4); // 在主线程，初始化子线程偏移位，所使用的共享内存
@State result: string = "";
build() {
Row() {
Column() {
// 不使用锁写入的按钮
Button($r('app.string.not_use_lock'))
.width("80%").fontSize(30)
.fontWeight(FontWeight.Bold)
.margin({ top: 30 })
.onClick(async () => {
this.startWrite(false);
})
// 使用锁写入的按钮
Button($r('app.string.use_lock'))
.width("80%")
.fontSize(30)
.fontWeight(FontWeight.Bold)
.margin({ top: 30 })
.onClick(async () => {
this.startWrite(true);
})
// 运行状态说明
Text(this.result)
.width("80%")
.fontSize(30)
.fontWeight(FontWeight.Bold)
.fontColor(Color.Blue)
.margin({ top: 30 })
}
.width('100%')
}
.height('100%')
}
startWrite(useLock: boolean): void {
// 指明运行状态为“写入文件开始”
this.result = getContext().resourceManager.getStringSync($r('app.string.write_file_start'));
// 初始化写入时的偏移量
let whichLineToWrite: Int32Array = new Int32Array(this.sabForLine);
Atomics.store(whichLineToWrite, 0, 0);
// 开启多线程依据偏移量指定位置写入文件
// 通过主线程的sabInLock：SharedArrayBuffer初始化锁，保证多线程操作同一处锁标志位
// 通过主线程的sabForLine：SharedArrayBuffer初始化偏移位，保证多线程操作同一处偏移位置
let taskPoolGroup: taskpool.TaskGroup = new taskpool.TaskGroup();
for (let i: number = 0; i < this.taskNum; i++) {
taskPoolGroup.addTask(new taskpool.Task(createWriteTask, this.baseDir, i, this.sabInLock, this.sabForLine, useLock));
}
taskpool.execute(taskPoolGroup).then(() => {
// 指明运行状态为“写入文件成功”
this.result = this.result = getContext().resourceManager.getStringSync($r('app.string.write_file_success'));
}).catch(() => {
// 指明运行状态为“写入文件失败”
this.result = getContext().resourceManager.getStringSync($r('app.string.write_file_failed'));
})
}
}
子线程根据偏移量在指定位置写入文件，并通过偏移量自增，指定下次的写入位置。
@Concurrent
async function createWriteTask(baseDir: string, writeText: number, sabInLock: SharedArrayBuffer, sabForLine: SharedArrayBuffer, useLock: boolean): Promise<void> {
class Option { // 写入文件时的接口方法参数类
offset: number = 0;
length: number = 0;
encoding: string = 'utf-8';
constructor(offset: number, length: number) {
this.offset = offset;
this.length = length;
}
}
// 初始化输出文件目录
let filePath: string | undefined = undefined;
filePath = baseDir + useLock ? "/useLock.txt" : "/unusedLock.txt";
if (!fs.accessSync(baseDir)) {
fs.mkdirSync(baseDir);
}
// 利用主线程传入的SharedArrayBuffer初始化锁
let nrl: NonReentrantLock | undefined = undefined;
if (useLock) {
nrl = new NonReentrantLock(sabInLock);
}
// 利用主线程传入的SharedArrayBuffer初始化写入文件时的偏移量
let whichLineToWrite: Int32Array = new Int32Array(sabForLine);
let str: string = writeText + '\n';
for (let i: number = 0; i < 100; i++) {
// 获取锁
if (useLock && nrl !== undefined) {
nrl.lock();
}
// 写入文件
let file: fs.File = fs.openSync(filePath, fs.OpenMode.READ_WRITE | fs.OpenMode.CREATE);
try {
fs.writeSync(file.fd, str, new Option(whichLineToWrite[0], str.length));
} catch (err) {
logger.error(`errorCode : ${err.code},errMessage : ${err.message}`);
}
fs.closeSync(file);
// 修改偏移量，指定下次写入时的位置
whichLineToWrite[0] += str.length;
// 释放锁
if (useLock && nrl !== undefined) {
nrl.unlock();
}
}
}
从应用沙箱地址查看写入的文件，可以看到unusedLock.txt文件，所写行数不足1000行，且存在乱码，如图1所示。
图1 不使用锁写入的文件
而usedLock.txt文件，所写行数刚好1000行，且不存在乱码，如图2所示。
图2 使用锁写入的文件
总结
综上所述，虽然使用了基于消息通信的Actor并发模型，但是ArkTS依旧支持通过共享内存的方式进行线程间通信。同时，在使用SharedArrayBuffer进行共享内存时，也需要通过原子操作或者锁来解决线程间同步与互斥的问题。合理使用多线程共享内存，才能在保证线程安全的前提下，提升应用的性能。