鸿蒙（HarmonyOS）性能优化实战-多线程共享内存

概述

在应用开发中，为了避免主线程阻塞，提高应用性能，需要将一些耗时操作放在子线程中执行。此时，子线程就需要访问主线程中的数据。ArkTS采用了基于消息通信的Actor并发模型，具有内存隔离的特性，所以跨线程传输数据时需要将数据序列化，但是AkrTS支持通过可共享对象SharedArrayBuffer实现直接的共享内存。

在开发应用时，如果遇到数据量较大，并且需要多个线程同时操作的情况，推荐使用SharedArrayBuffer共享内存，可以减少数据在线程间传递时需要复制和序列化的额外开销。比如，音视频解码播放、多个线程同时读取写入文件等场景。由于内存是共享的，所以在多个线程同时操作同一块内存时，可能会引起数据的紊乱，这时就需要使用锁来确保数据操作的有序性。本文将基于此具体展开说明。关于多线程的使用和原理，可参考OpenHarmony多线程能力场景化示例实践，本文将不再详细讲述。

工作原理

可共享对象SharedArrayBuffer，是拥有固定长度的原始二进制数据缓冲区，可以存储任何类型的数据，包括数字、字符串等。它支持在多线程之间传递，传递之后的SharedArrayBuffer对象和原始的SharedArrayBuffer对象可以指向同一块内存，进而达到共享内存的目的。SharedArrayBuffer对象存储的数据在子线程中被修改时，需要通过原子操作保证其同步性，即下个操作开始之前务必需要保证上个操作已经结束。下面将通过示例说明原子操作保证同步性的必要性。

非原子操作

......
// 非原子操作，进行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的SharedArrayBufferthis.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();}}
}