HarmonyOS实战（解决方案篇）—从实战案例了解应用并发设计

HarmonyOS实战（解决方案篇）—从实战案例了解应用并发设计

• • 引言
  • 一、并发能力整体架构
  • • 并发能力框架
    • ArkTS并发模型与业界模型的差异
    • TaskPool与Worker对比
  • 二、AI证件照工具并发改造实战
  • • 2.1 原始代码的问题分析
    • 2.2 方案一：TaskPool处理耗时预处理
    • • 什么是TaskPool？
      • 改造代码
      • 改造效果
    • 2.3 方案二：Worker处理AI推理长时任务
    • • 什么是Worker？
      • 创建Worker
      • Worker的优势
    • 2.4 方案三：TaskGroup批量处理多张照片
    • • 什么是TaskGroup？
      • 批量处理实现
      • TaskGroup的优势
    • 2.5 方案四：序列任务处理
    • • 使用SequenceRunner实现串行执行
    • 2.6 方案五：线程间通信优化
    • • 使用SharedArrayBuffer共享内存
      • 使用Sendable对象共享模型实例
    • 2.7 方案六：生产者-消费者模式实现预览队列
  • 三、性能对比与最佳实践
  • • 3.1 改造前后性能对比
    • 3.2 并发选型指南
    • 3.3 最佳实践总结
  • 四、总结与展望
  • 参考资料

引言

大家好，我是你们的老朋友木斯佳。首先祝大家马年大吉，身体健康，万事如意！给大家拜个晚年啦~

假期虽然马上结束了，但学习和分享的脚步不能停。今天要和大家聊的是HarmonyOS应用开发中的核心话题——并发设计。

不知道大家在实际开发中是否遇到过这样的困扰：应用界面卡顿、掉帧，用户体验不佳？或者明明功能都实现了，但总觉得性能差强人意？很多时候，这些问题的根源很可能就在于并发设计不合理。

在之前的深色模式适配文章中，我们通过AI助手这个案例，详细讲解了如何让应用在不同主题下都有出色的视觉效果。这次，我们将继续以之前开发过的应用为例，深入剖析它在并发设计上的实战经验。

通过这些真实案例，我们将系统讲解：

✅ 耗时任务、长时任务、常驻任务如何选择TaskPool还是Worker ✅ 多任务串行执行、依赖执行、批量执行的实现技巧 ✅ 线程间高效通信的最佳实践 ✅ 单例模式在并发环境下的正确姿势 ✅ 生产者-消费者模式的经典应用

无论你是刚开始接触HarmonyOS并发开发的新手，还是希望优化现有应用性能的老手，这篇文章都能给你带来实用的参考价值。

话不多说，让我们开始今天的实战之旅吧！🚀

一、并发能力整体架构

并发能力框架

HarmonyOS的并发能力框架如下图所示：

• 主线程：执行UI业务、不耗时操作、单次I/O任务，与其他ArkTS线程共享系统I/O线程池
• TaskPool高并发任务池：执行耗时任务，封装任务入口，统计模块负载，开发者无需管理线程生命周期
• Worker线程：执行常驻任务，CPU密集型、耗时任务，限制线程个数为64
• FFRT任务池：系统任务和用户C/C++耗时任务的调度池
• Pthread线程：C/C++开发的模块，后台运行或耗时的ArkTS无关业务，不限制线程个数

ArkTS并发模型与业界模型的差异

传统共享内存并发模型：

采用线程和锁的并发机制，不同线程共享内存并通过锁保护临界区。对于包含I/O操作或锁的业务，为防止阻塞，需开启多个线程执行不同业务，导致应用经常存在几百个线程，增加调度开销和内存占用。

ArkTS并发模型：

采用内存隔离的线程模型，不同线程间通过消息通信，线程内无锁化运行。业务内部的I/O操作由系统分发到后台的I/O任务池，不阻塞ArkTS上层逻辑。异步I/O不阻塞ArkTS线程，TaskPool及I/O线程池由系统统一管理，大幅提升能效。

TaskPool与Worker对比

二、AI证件照工具并发改造实战

2.1 原始代码的问题分析

先来看一下我们最初实现的证件照处理代码，感受一下什么叫"卡到怀疑人生"：

// ❌ 错误示例：所有操作都在主线程执行
async function processIDPhoto(imageUri: string): Promise<void> {
// 显示加载中
this.isProcessing = true;

// 1. 图像预处理 – 循环像素操作，主线程卡顿
const pixelMap = await loadAndPreprocessImage(imageUri);

// 2. AI推理 – 2-3秒的CPU密集操作，界面完全无响应
const mask = await runMindSporeInference(pixelMap);

// 3. 后处理生成证件照
const result = await generateIDPhoto(pixelMap, mask);

// 4. 更新UI
this.showResult(result);
this.isProcessing = false;
}

这段代码的问题在于：所有耗时操作都在主线程执行。当用户点击"开始处理"按钮后，整个UI线程被阻塞，无法响应用户的任何操作。如果处理时间超过5秒，系统甚至会弹出"应用无响应"的提示。

2.2 方案一：TaskPool处理耗时预处理

对于图像预处理这类独立、耗时短的任务，TaskPool是最佳选择。

什么是TaskPool？

TaskPool是HarmonyOS提供的高并发任务池，开发者只需将任务封装好交给TaskPool，系统自动管理线程的创建、调度和销毁。

改造代码

首先，将图像预处理逻辑封装成一个并发函数：

// imagePreprocess.ets

// 1. 使用@Concurrent装饰器标记这是一个可并发执行的函数
@Concurrent
export async function preprocessImage(imageUri: string): Promise<ArrayBuffer> {
console.info('🚀 [TaskPool] 开始图像预处理');

// 打开文件
let file = fileIo.openSync(imageUri, fileIo.OpenMode.READ_ONLY);
let imageSource = image.createImageSource(file.fd);
let pixelMap = await imageSource.createPixelMapSync();

// 获取原始尺寸
const { width: originalWidth, height: originalHeight } = pixelMap.getImageInfoSync();
console.info(`📐 原始尺寸: ${originalWidth}x${originalHeight}`);

// 缩放到模型输入尺寸 (1024×1024)
pixelMap.scaleSync(1024 / originalWidth, 1024 / originalHeight);

// 读取像素数据
let readBuffer = new ArrayBuffer(1024 * 1024 * 4);
await pixelMap.readPixelsToBuffer(readBuffer);
const imageArr = new Uint8Array(readBuffer);

// 归一化处理
let float32View = new Float32Array(1024 * 1024 * 3);
let means = [0.5, 0.5, 0.5];
let stds = [1.0, 1.0, 1.0];

let index = 0;
for (let i = 0; i < imageArr.length; i++) {
if ((i + 1) % 4 === 0) {
float32View[index] = (imageArr[i – 3] / 255.0 – means[0]) / stds[0];
float32View[index + 1] = (imageArr[i – 2] / 255.0 – means[1]) / stds[1];
float32View[index + 2] = (imageArr[i – 1] / 255.0 – means[2]) / stds[2];
index += 3;
}
}

// 释放资源
pixelMap.release();
imageSource.release();
fileIo.closeSync(file.fd);

console.info('✅ [TaskPool] 预处理完成');
return float32View.buffer;
}

然后，在主线程中调用TaskPool执行：

// IDPhotoProcessor.ets
//…

async function processWithTaskPool(imageUri: string) {
this.isProcessing = true;

try {
// 2. 使用taskpool.execute执行并发任务
// 第一个参数是并发函数，后面是传给该函数的参数
const inputBuffer = await taskpool.execute(preprocessImage, imageUri)
.then((result: Object) => {
console.info('✅ 预处理任务完成');
return result as ArrayBuffer;
})
.catch((err: BusinessError) => {
console.error(`预处理失败: ${err.message}`);
throw err;
});

// 接下来进行AI推理…
await this.runInference(inputBuffer);

} catch (error) {
console.error('处理失败:', error);
} finally {
this.isProcessing = false;
}
}

改造效果

经过TaskPool改造后，主线程不再被阻塞！用户点击处理按钮后，UI仍然可以响应用户操作，比如取消处理、调整参数等。虽然预处理任务仍在后台执行，但界面再也不会"假死"了。

2.3 方案二：Worker处理AI推理长时任务

对于AI推理这种执行时间较长（>3分钟） 的任务，TaskPool就不太合适了。因为TaskPool中的任务如果执行时间过长，会被系统回收。这时需要使用Worker。

什么是Worker？

Worker是HarmonyOS提供的独立线程解决方案，开发者可以创建长期运行的Worker线程，并手动控制其生命周期。Worker适合执行常驻任务或长耗时任务。

创建Worker

首先，创建Worker文件 entry/src/main/ets/workers/AIInferenceWorker.ets（篇幅问题略过）

然后主线程中使用Worker

// IDPhotoProcessor.ets
export class IDPhotoProcessor {
private workerInstance: worker.ThreadWorker | null = null;
private taskCallbacks: Map<string, { resolve: Function, reject: Function }> = new Map();

// 初始化Worker
async initWorker(): Promise<void> {
if (this.workerInstance) return;

return new Promise((resolve, reject) => {
// 创建Worker实例
this.workerInstance = new worker.ThreadWorker(
'entry/ets/workers/AIInferenceWorker.ets',
{ name: 'AI Inference Worker' }
);

const taskId = 'init-' + Date.now();
this.taskCallbacks.set(taskId, { resolve, reject });

// 监听Worker消息
this.workerInstance.onmessage = (event: worker.MessageEvents) => {
const { type, data, error, taskId } = event.data;
const callback = this.taskCallbacks.get(taskId);

if (!callback) return;

if (type === 'initDone') {
callback.resolve();
this.taskCallbacks.delete(taskId);
} else if (type === 'error') {
callback.reject(new Error(error));
this.taskCallbacks.delete(taskId);
} else if (type === 'inferenceResult') {
callback.resolve(data);
this.taskCallbacks.delete(taskId);
}
};

// 加载模型文件
const context = getContext(this) as common.UIAbilityContext;
const resMgr = context.resourceManager;
const modelBuffer = resMgr.getRawFileContentSync(MODEL_NAME);

// 发送初始化消息给Worker
this.workerInstance.postMessage({
type: 'init',
data: { modelBuffer: modelBuffer.buffer },
taskId
});
});
}

// 执行AI推理
async runInference(inputBuffer: ArrayBuffer): Promise<ArrayBuffer> {
if (!this.workerInstance) {
await this.initWorker();
}

return new Promise((resolve, reject) => {
const taskId = 'inference-' + Date.now() + '-' + Math.random();
this.taskCallbacks.set(taskId, { resolve, reject });

this.workerInstance!.postMessage({
type: 'inference',
data: { inputBuffer },
taskId
});
});
}

// 释放Worker资源
releaseWorker(): void {
if (this.workerInstance) {
this.workerInstance.postMessage({ type: 'release' });
this.workerInstance.terminate();
this.workerInstance = null;
}
}
}

Worker的优势

通过Worker改造AI推理后，我们获得了几个关键能力：

2.4 方案三：TaskGroup批量处理多张照片

用户经常需要一次性生成多种底色（红底、蓝底、白底）的证件照。如果串行处理，需要等待很长时间；如果并行处理，又需要合理管理多个任务。

什么是TaskGroup？

TaskGroup是TaskPool提供的任务组功能，可以将多个任务加入一个组，统一等待所有任务完成。

批量处理实现

在实际工程上实现要比下面代码复杂，建议抽取独立的工具类，我们在之前的博客中对于这部分代码有比较详细的介绍，下面的代码作为伪代码提供执行思路。

// BatchProcessor.ets

@Concurrent
async function generateColorBackground(
imageData: ArrayBuffer,
maskData: ArrayBuffer,
bgColor: [number, number, number]
): Promise<ArrayBuffer> {
console.info(`🎨 生成${bgColor}背景证件照`);

// 将原始图像和分割掩码合成为指定背景色的证件照
const imageArr = new Uint8Array(imageData);
const maskArr = new Uint8Array(maskData);
const result = new Uint8Array(1024 * 1024 * 4);

for (let i = 0; i < result.length; i += 4) {
const maskAlpha = maskArr[i + 3] / 255.0;

if (maskAlpha > 0.5) {
// 人像区域保留原色

} else {
// 背景区域替换为指定颜色

}
}

return result.buffer;
}

async function batchGenerateIDPhotos(
imageData: ArrayBuffer,
maskData: ArrayBuffer
): Promise<void> {
// 定义需要生成的背景色
const bgColors: Array<[string, [number, number, number]]> = [
//…….
];

// 创建任务组
let taskGroup = new taskpool.TaskGroup();
let tasks: taskpool.Task[] = [];

// 为每种背景色创建任务
bgColors.forEach(([name, color]) => {
let task = new taskpool.Task(generateColorBackground, imageData, maskData, color);
tasks.push(task);
taskGroup.addTask(task);
console.info(`📋 添加任务: ${name}`);
});

try {
// 执行任务组，等待所有任务完成
console.info('🚀 开始批量生成…');
const results = await taskpool.execute(taskGroup) as ArrayBuffer[];

// 处理结果
results.forEach((result, index) => {
const [name] = bgColors[index];
console.info(`✅ ${name}生成完成`);
// 保存或显示结果
this.saveResult(result, name);
});

console.info('🎉 所有证件照生成完成');
} catch (error) {
console.error('批量生成失败:', error);
}
}

TaskGroup的优势

2.5 方案四：序列任务处理

在某些场景下，任务需要按特定顺序执行。比如证件照处理流程：

这些步骤有严格的依赖关系，不能并发执行。

使用SequenceRunner实现串行执行

// SerialProcessor.ets
import { taskpool } from '@kit.ArkTS';

@Concurrent
async function detectFace(imageUri: string): Promise<{ hasFace: boolean, faceRect?: Rect }> {
console.info('🔍 检测人脸…');
// 人脸检测逻辑
return { hasFace: true, faceRect: { x: 100, y: 100, width: 200, height: 200 } };
}

@Concurrent
async function segmentPortrait(imageUri: string, faceRect: Rect): Promise<ArrayBuffer> {
console.info('✂️ 人像分割…');
// 人像分割逻辑
return new ArrayBuffer(1024 * 1024 * 4);
}

@Concurrent
async function generateIDPhotoFromMask(imageUri: string, mask: ArrayBuffer, bgColor: string): Promise<ArrayBuffer> {
console.info('📸 生成证件照…');
// 证件照生成逻辑
return new ArrayBuffer(1024 * 1024 * 4);
}

async function processIDPhotoSerial(imageUri: string, bgColor: string): Promise<void> {
// 创建串行执行器
const runner = new taskpool.SequenceRunner();

// 创建任务
const faceTask = new taskpool.Task(detectFace, imageUri);
let faceRect: Rect | undefined;

// 按顺序执行任务
try {
// 第一步：人脸检测
await runner.execute(faceTask).then((result: any) => {
if (!result.hasFace) {
throw new Error('未检测到人脸');
}
faceRect = result.faceRect;
console.info('✅ 人脸检测完成');
});

// 第二步：人像分割（依赖人脸检测结果）
const segmentTask = new taskpool.Task(segmentPortrait, imageUri, faceRect);
let mask: ArrayBuffer;
await runner.execute(segmentTask).then((result: ArrayBuffer) => {
mask = result;
console.info('✅ 人像分割完成');
});

// 第三步：生成证件照（依赖分割结果）
const generateTask = new taskpool.Task(generateIDPhotoFromMask, imageUri, mask, bgColor);
await runner.execute(generateTask).then((result: ArrayBuffer) => {
console.info('✅ 证件照生成完成');
this.showResult(result);
});

} catch (error) {
console.error('处理失败:', error);
}
}

2.6 方案五：线程间通信优化

在AI证件照工具中，线程间需要传递大量数据：图像数据、分割掩码等。如果不加优化，频繁的大对象拷贝会成为性能瓶颈。

使用SharedArrayBuffer共享内存

对于需要在多个线程间共享的数据，可以使用SharedArrayBuffer：

// SharedMemoryProcessor.ets
import { taskpool } from '@kit.ArkTS';

@Concurrent
function processWithSharedMemory(sharedBuffer: SharedArrayBuffer, width: number, height: number): void {
// 将SharedArrayBuffer包装为Uint8Array进行操作
const pixels = new Uint8Array(sharedBuffer);

// 直接在共享内存上操作，不需要拷贝
for (let i = 0; i < pixels.length; i += 4) {
// 处理像素…
// 修改直接反映在共享内存中
}

console.info('✅ 共享内存处理完成');
}

async function useSharedMemory(imageUri: string): Promise<void> {
// 加载图像
let pixelMap = await loadImage(imageUri);
const { width, height } = pixelMap.getImageInfoSync();

// 创建共享内存
const sharedBuffer = new SharedArrayBuffer(width * height * 4);
let buffer = new Uint8Array(sharedBuffer);

// 将像素数据读入共享内存
await pixelMap.readPixelsToBuffer(sharedBuffer);

// 将共享内存传递给TaskPool任务
const task = new taskpool.Task(processWithSharedMemory, sharedBuffer, width, height);
await taskpool.execute(task);

// 读取处理后的数据（直接从共享内存获取）
const processedPixels = new Uint8Array(sharedBuffer);

// 创建新的PixelMap显示结果
const resultPixelMap = await image.createPixelMapFromData(processedPixels, {
size: { width, height }
});

this.showResult(resultPixelMap);
}

使用Sendable对象共享模型实例

对于需要在线程间共享的复杂对象，可以定义为Sendable类：

// sendable/ModelLoader.ets
"use shared"

@Sendable
export class ModelLoader {
private static instance: ModelLoader;
private modelBuffer: ArrayBuffer | null = null;

private constructor() {}

public static getInstance(): ModelLoader {
if (!ModelLoader.instance) {
ModelLoader.instance = new ModelLoader();
}
return ModelLoader.instance;
}

public async loadModel(context: common.UIAbilityContext): Promise<void> {
if (this.modelBuffer) return;

const resMgr = context.resourceManager;
this.modelBuffer = (await resMgr.getRawFileContent(MODEL_NAME)).buffer;
console.info('✅ 模型加载完成');
}

public getModelBuffer(): ArrayBuffer {
if (!this.modelBuffer) {
throw new Error('模型未加载');
}
return this.modelBuffer;
}
}

然后在多个线程中使用：

// 主线程
import { ModelLoader } from './sendable/ModelLoader';

async function initModel() {
const loader = ModelLoader.getInstance();
await loader.loadModel(getContext(this));
}

// Worker线程
import { ModelLoader } from '../sendable/ModelLoader';
import { worker } from '@kit.ArkTS';

const workerPort = worker.workerPort;
workerPort.onmessage = async () => {
// 直接获取单例实例
const loader = ModelLoader.getInstance();
const modelBuffer = loader.getModelBuffer();
// 使用模型…
};

2.7 方案六：生产者-消费者模式实现预览队列

对于实时预览场景，我们需要一个高效的生产者-消费者模式来处理视频帧。

// PreviewProcessor.ets
import { taskpool } from '@kit.ArkTS';

// 帧队列
class FrameQueue {
private queue: Array<{ frameData: ArrayBuffer, timestamp: number }> = [];
private maxSize: number = 5;
private processing = false;

// 生产者：添加帧
async addFrame(frameData: ArrayBuffer, timestamp: number): Promise<void> {
if (this.queue.length >= this.maxSize) {
// 队列满时丢弃最旧的帧
this.queue.shift();
}

this.queue.push({ frameData, timestamp });

if (!this.processing) {
this.processQueue();
}
}

// 消费者：处理队列
private async processQueue(): Promise<void> {
if (this.queue.length === 0) {
this.processing = false;
return;
}

this.processing = true;
const frame = this.queue.shift()!;

try {
// 将帧处理任务交给TaskPool
const result = await taskpool.execute(processFrame, frame.frameData);

// 在主线程更新预览
this.updatePreview(result, frame.timestamp);

} catch (error) {
console.error('帧处理失败:', error);
}

// 继续处理下一帧
this.processQueue();
}

@Concurrent
private static async processFrame(frameData: ArrayBuffer): Promise<ArrayBuffer> {
// 轻量级推理（使用更小的模型或降低分辨率）
// …
return processedData;
}
}

// 摄像头预览使用
class CameraPreview {
private frameQueue = new FrameQueue();

// 摄像头帧回调（高频调用）
onFrame(frameData: ArrayBuffer, timestamp: number): void {
// 直接入队，不阻塞
this.frameQueue.addFrame(frameData, timestamp);
}
}

三、性能对比与最佳实践

3.1 改造前后性能对比

*注：通过共享内存优化，实际内存增长有限

3.2 并发选型指南

3.3 最佳实践总结

四、总结与展望

通过AI证件照工具的真实案例，我们系统学习了HarmonyOS的并发设计：

• TaskPool 处理图像预处理等独立耗时任务
• Worker 处理AI推理等长时任务
• TaskGroup 实现批量并行处理
• SequenceRunner 保证任务串行执行
• SharedArrayBuffer 优化大对象共享
• Sendable 实现线程安全单例
• 生产者-消费者模式 处理实时预览

未来，随着HarmonyOS并发能力的不断增强，我们还可以探索更多可能：

• 使用NPU硬件加速AI推理
• 实现更复杂的多模态输入处理
• 支持分布式设备协同计算

端侧AI + 高效并发，正在让移动应用变得更加智能和流畅。希望本文能帮助你在自己的项目中，设计出更优秀的并发方案。

参考资料