鸿蒙实战｜并发编程深度解析：TaskPool、Worker 与 ArkTS 并发模型
·
一、我们要做什么
1.1 业务场景需求
| 场景 | 操作类型 | 数据规模 | 并发价值 |
|---|---|---|---|
| 图片批量处理 | CPU 密集 | 100+ 张 | 极高 |
| JSON 大文件解析 | CPU 密集 | 10MB+ | 高 |
| 音视频转码 | CPU 密集 | 持续流 | 极高 |
| 多接口数据聚合 | I/O 密集 | 多接口 | 高 |
| 大数据集排序 | CPU 密集 | 10 万+ | 极高 |
| 定时任务调度 | 混合 | 不限 | 中 |
1.2 预期效果
引入 TaskPool 和 Worker 后，目标如下：CPU 密集任务耗时降至单线程的 30%~50%；主线程不被阻塞，UI 保持 60fps 流畅；Sendable 对象使数据传递效率提升 70% 以上。
1.3 技术挑战
ArkTS 并发基于 Actor 内存隔离模型，线程间不能直接共享引用，跨线程传递必须经过序列化。开发者需清晰区分 Sendable 对象与普通对象。TaskPool 适合短时轻量任务，Worker 适合长时重计算，选型错误反而增加开销。
二、数据模型设计
2.1 任务模型定义
// TaskModel.ets
export class TaskConfig {
id: string;
type: TaskType;
priority: TaskPriority;
timeout: number;
retries: number;
constructor(id: string, type: TaskType, priority: TaskPriority = TaskPriority.MEDIUM, timeout: number = 30000) {
this.id = id; this.type = type; this.priority = priority;
this.timeout = timeout; this.retries = 0;
}
}
export enum TaskType {
IMAGE_PROCESS = 'image_process',
DATA_PARSE = 'data_parse',
COMPUTATION = 'computation'
}
export enum TaskPriority {
LOW = 0, MEDIUM = 1, HIGH = 2, EXCLUSIVE = 3
}
export class TaskResult {
taskId: string; success: boolean; data: Object; errorMessage: string; duration: number;
constructor(taskId: string) { this.taskId = taskId; this.success = false; this.data = {}; this.errorMessage = ''; this.duration = 0; }
}
export class TaskStats {
totalSubmitted: number = 0; totalCompleted: number = 0; totalFailed: number = 0;
}
2.2 Sendable 数据契约
@Sendable 装饰的类实例可直接跨线程传递，无需手动序列化。属性只能是基础类型、ArrayBuffer、Sendable 类或其嵌套结构，不允许包含闭包或函数引用。
// SendableData.ets
@Sendable
export class ImageTaskData {
imagePath: string; width: number; height: number; filterType: string; outputPath: string;
constructor(path: string, w: number, h: number, filter: string) {
this.imagePath = path; this.width = w; this.height = h; this.filterType = filter; this.outputPath = '';
}
}
@Sendable
export class ComputeTaskData {
dataset: ArrayBuffer; algorithmType: string; params: Record<string, number>;
constructor(buffer: ArrayBuffer, algo: string, params: Record<string, number>) {
this.dataset = buffer; this.algorithmType = algo; this.params = params;
}
}
2.3 消息协议设计
Worker 线程与主线程之间通过 postMessage 和 onmessage 异步通信：
// MessageProtocol.ets
export enum MessageType { INIT = 'init', EXECUTE = 'execute', RESULT = 'result', ERROR = 'error', PROGRESS = 'progress', CANCEL = 'cancel' }
@Sendable
export class WorkerMessage {
type: MessageType; payload: Object; requestId: string; timestamp: number;
constructor(type: MessageType, payload: Object, requestId: string) {
this.type = type; this.payload = payload; this.requestId = requestId; this.timestamp = Date.now();
}
}
三、核心设计决策
3.1 TaskPool 与 Worker 选型对比
| 维度 | TaskPool | Worker |
|---|---|---|
| 线程管理 | 系统自动管理线程池 | 开发者手动创建销毁 |
| 适用场景 | 短时、频繁、轻量级 | 耗时较长、长期运行 |
| 生命周期 | 任务结束自动释放 | 与宿主线程绑定，需手动 terminate |
| 优先级 | 支持 taskPriority | 不支持 |
| 任务上限 | 约 1000 个槽位 | 仅受内存限制 |
| 数据传递 | @Sendable 对象 | postMessage（自动序列化） |
| 错误处理 | try-catch 回调包装 | onerror 事件监听 |
选型建议：优先 TaskPool。出现以下情况切换 Worker——任务超过 30 秒、需要独立线程上下文、需要多 Worker 相互通信、或任务数远超 TaskPool 槽位上限。
3.2 @Concurrent 与普通异步函数对比
| 维度 | @Concurrent 函数 | 普通异步函数 |
|---|---|---|
| 执行线程 | 后台线程池 | 主线程或其他线程 |
| 阻塞主线程 | 不阻塞 | 不阻塞 |
| 调用方式 | taskpool.execute(函数, 参数) | 函数().then() |
| 类型检查 | 编译时检查 Sendable | 无特殊要求 |
| 适用场景 | CPU 密集计算 | I/O 密集操作 |
3.3 Sendable 设计原则
跨线程数据传递是并发编程中最易出错的环节。设计原则：数据模型宁可拆分也不嵌套过深；ArrayBuffer 传二进制数据比 string 序列化效率高 3~5 倍；避免 Sendable 对象包含可变嵌套引用。
// ❌ 错误：闭包不允许作为 Sendable
export class NonSendableData { name: string; callback: Function; }
// ✅ 正确：符合 Sendable 契约
@Sendable export class ValidData { name: string; items: Array<number>; buffer: ArrayBuffer; }
四、完整代码实现
4.1 @Concurrent 并发函数实现
@Concurrent 函数必须是顶层函数，参数和返回值都必须是 Sendable 类型：
// concurrent_functions.ets — @Concurrent 顶层函数（必须在独立文件）
import { TaskResult } from '../model/TaskModel';
// 图像滤镜处理 — CPU 密集型
@Concurrent
export function processImageFilter(taskId: string, imagePath: string, filterType: string): TaskResult {
const result = new TaskResult(taskId);
const start = Date.now();
try {
let sum = 0;
for (let i = 0; i < 500000; i++) {
sum += Math.sqrt(i) * Math.sin(i * 0.01);
}
result.success = true;
result.data = { outputPath: imagePath.replace('.jpg', `_${filterType}_processed.jpg`), sumResult: sum };
result.duration = Date.now() - start;
} catch (e) {
result.success = false; result.errorMessage = (e as Error).message; result.duration = Date.now() - start;
}
return result;
}
// 大数据集快速排序
@Concurrent
export function sortLargeDataset(taskId: string, numbers: Array<number>, algorithm: string): TaskResult {
const result = new TaskResult(taskId);
const start = Date.now();
try {
const arr = [...numbers];
const sorted = algorithm === 'merge' ? mergeSort(arr) : quickSort(arr);
result.success = true;
result.data = { sortedData: sorted, originalLength: numbers.length, isSorted: isSorted(sorted) };
result.duration = Date.now() - start;
} catch (e) {
result.success = false; result.errorMessage = (e as Error).message; result.duration = Date.now() - start;
}
return result;
}
function quickSort(arr: Array<number>): Array<number> {
if (arr.length <= 1) return arr;
const pivot = arr[Math.floor(arr.length / 2)];
return [...quickSort(arr.filter(x => x < pivot)), ...arr.filter(x => x === pivot), ...quickSort(arr.filter(x => x > pivot))];
}
function mergeSort(arr: Array<number>): Array<number> {
if (arr.length <= 1) return arr;
const mid = Math.floor(arr.length / 2);
const merge = (l: number[], r: number[]): number[] => {
const res: number[] = []; let i = 0, j = 0;
while (i < l.length && j < r.length) res.push(l[i] <= r[j] ? l[i++] : r[j++]);
return res.concat(l.slice(i)).concat(r.slice(j));
};
return merge(mergeSort(arr.slice(0, mid)), mergeSort(arr.slice(mid)));
}
function isSorted(arr: Array<number>): boolean {
for (let i = 1; i < arr.length; i++) if (arr[i] < arr[i - 1]) return false;
return true;
}
4.2 TaskPool 任务池完整封装
TaskPoolManager 提供任务提交、分组、取消、超时处理和统计监控能力：
// TaskPoolManager.ets
import taskpool from '@ohos.taskpool';
import { TaskConfig, TaskResult, TaskPriority, TaskType, TaskStats } from '../model/TaskModel';
interface TaskRecord {
config: TaskConfig; task: taskpool.Task; submitTime: number;
status: 'pending' | 'running' | 'completed' | 'failed' | 'cancelled'; result?: TaskResult;
}
export class TaskPoolManager {
private static instance: TaskPoolManager;
private taskRecords: Map<string, TaskRecord> = new Map();
private taskGroups: Map<string, { name: string; taskIds: Set<string>; pendingCount: number }> = new Map();
private stats: TaskStats = new TaskStats();
private constructor() {}
static getInstance(): TaskPoolManager {
if (!TaskPoolManager.instance) TaskPoolManager.instance = new TaskPoolManager();
return TaskPoolManager.instance;
}
createGroup(groupName: string): void {
this.taskGroups.set(groupName, { name: groupName, taskIds: new Set(), pendingCount: 0 });
}
async submitTask(groupName: string, func: Function, args: Array<Object>, config: TaskConfig): Promise<TaskResult> {
const startTime = Date.now();
const task = new taskpool.Task(func.name, func, args);
// 设置优先级
task.priority = config.priority === TaskPriority.HIGH ? taskpool.Priority.HIGH
: config.priority === TaskPriority.LOW ? taskpool.Priority.LOW : taskpool.Priority.MEDIUM;
const record: TaskRecord = { config, task, submitTime: startTime, status: 'pending' };
this.taskRecords.set(config.id, record);
const group = this.taskGroups.get(groupName);
if (group) { group.taskIds.add(config.id); group.pendingCount++; }
this.stats.totalSubmitted++;
try {
const result = config.timeout > 0
? await Promise.race([
taskpool.execute(task) as Promise<TaskResult>,
new Promise<never>((_, reject) => setTimeout(() => {
taskpool.cancel(task);
reject(new Error(`Task ${config.id} timed out after ${config.timeout}ms`));
}, config.timeout))
])
: await taskpool.execute(task) as TaskResult;
record.status = 'completed'; record.result = result; this.stats.totalCompleted++;
if (group) group.pendingCount = Math.max(0, group.pendingCount - 1);
return result;
} catch (e) {
record.status = 'failed';
record.result = Object.assign(new TaskResult(config.id), { success: false, errorMessage: (e as Error).message, duration: Date.now() - startTime });
this.stats.totalFailed++;
if (group) group.pendingCount = Math.max(0, group.pendingCount - 1);
return record.result;
}
}
async submitBatch(groupName: string, tasks: Array<{ func: Function; args: Array<Object>; config: TaskConfig }>): Promise<TaskResult[]> {
return Promise.all(tasks.map(t => this.submitTask(groupName, t.func, t.args, t.config)));
}
cancelTask(taskId: string): boolean {
const record = this.taskRecords.get(taskId);
if (!record || record.status === 'completed') return false;
const cancelled = taskpool.cancel(record.task);
if (cancelled) record.status = 'cancelled';
return cancelled;
}
cancelGroup(groupName: string): number {
const group = this.taskGroups.get(groupName);
if (!group) return 0;
let count = 0;
group.taskIds.forEach(id => { if (this.cancelTask(id)) count++; });
return count;
}
getGroupStatus(groupName: string): { total: number; pending: number; completed: number } {
const group = this.taskGroups.get(groupName);
if (!group) return { total: 0, pending: 0, completed: 0 };
let completed = 0;
group.taskIds.forEach(id => { const r = this.taskRecords.get(id); if (r && r.status === 'completed') completed++; });
return { total: group.taskIds.size, pending: group.pendingCount, completed };
}
getStats(): TaskStats { return { ...this.stats }; }
resetStats(): void { this.stats = new TaskStats(); }
}
4.3 Worker 多线程实现
Worker 适合需要长期独立运行的计算密集型任务：
// workers/LongRunningWorker.ts — Worker 线程端实现
import { WorkerMessage, MessageType } from '../model/MessageProtocol';
let workerData: Object = {};
onmessage = (event: MessageEvent<WorkerMessage>) => {
const msg = event.data;
switch (msg.type) {
case MessageType.INIT: handleInit(msg); break;
case MessageType.EXECUTE: handleExecute(msg); break;
case MessageType.CANCEL: postResult(msg.requestId, { status: 'cancelled' }); break;
default: postError(msg.requestId, `Unknown type: ${msg.type}`);
}
};
function handleInit(msg: WorkerMessage) {
workerData = msg.payload || {};
postResult(msg.requestId, { status: 'initialized', workerId: msg.requestId, timestamp: Date.now() });
}
function handleExecute(msg: WorkerMessage) {
const { taskType, params } = msg.payload as { taskType: string; params: Object };
const startTime = Date.now();
try {
let result: Object = {};
switch (taskType) {
case 'video_decode': result = simulateVideoDecode(params); break;
case 'large_sort': result = simulateLargeSort(params); break;
case 'data_agg': result = simulateDataAgg(params); break;
default: throw new Error(`Unsupported: ${taskType}`);
}
postProgress(msg.requestId, 50, 'Processing...');
postResult(msg.requestId, { ...result, duration: Date.now() - startTime, workerId: msg.requestId });
} catch (e) { postError(msg.requestId, (e as Error).message); }
}
function simulateVideoDecode(params: Object): Object {
const { frameCount } = params as { frameCount: number };
const frames: ArrayBuffer[] = [];
for (let i = 0; i < Math.min(frameCount, 10); i++) frames.push(new ArrayBuffer(1920 * 1080 * 4));
return { decodedFrames: frames.length, totalSize: frames.reduce((s, f) => s + f.byteLength, 0) };
}
function simulateLargeSort(params: Object): Object {
const { recordCount } = params as { recordCount: number };
const arr: Array<number> = [];
for (let i = 0; i < Math.min(recordCount, 10000); i++) arr.push(Math.random() * 10000);
const sorted = arr.sort((a, b) => a - b);
return { recordCount: sorted.length, isSorted: true };
}
function simulateDataAgg(params: Object): Object {
const { sources } = params as { sources: number };
const aggregated: Record<string, number> = {};
for (let i = 0; i < sources; i++) aggregated[`source_${i}`] = Math.floor(Math.random() * 1000);
const total = Object.values(aggregated).reduce((s, v) => s + v, 0);
return { sourceCount: sources, totalValue: total, avgValue: total / sources };
}
function postResult(requestId: string, data: Object): void {
postMessage(new WorkerMessage(MessageType.RESULT, data, requestId));
}
function postError(requestId: string, error: string): void {
postMessage(new WorkerMessage(MessageType.ERROR, { error }, requestId));
}
function postProgress(requestId: string, percent: number, status: string): void {
postMessage(new WorkerMessage(MessageType.PROGRESS, { percent, status }, requestId));
}
// WorkerManager.ets — 主线程端 Worker 管理封装
import worker from '@ohos.worker';
import { WorkerMessage, MessageType } from '../model/MessageProtocol';
export type WorkerCallback = (result: Object, error?: string) => void;
export type ProgressCallback = (percent: number, status: string) => void;
export class WorkerManager {
private activeWorkers: Map<string, worker.ThreadWorker> = new Map();
private callbacks: Map<string, { onResult: WorkerCallback; onProgress?: ProgressCallback }> = new Map();
createWorker(workerPath: string, workerId: string): void {
if (this.activeWorkers.has(workerId)) return;
const w = new worker.ThreadWorker(workerPath, { name: workerId, sharedThreadMemory: 1024 * 1024 });
w.onmessage = (event: MessageEvent<WorkerMessage>) => this.handleMessage(workerId, event.data);
w.onerror = (e: Error) => { console.error(`Worker ${workerId} error: ${e.message}`); this.activeWorkers.delete(workerId); };
this.activeWorkers.set(workerId, w);
}
sendMessage(workerId: string, message: WorkerMessage, cb: { onResult: WorkerCallback; onProgress?: ProgressCallback }): void {
const w = this.activeWorkers.get(workerId);
if (!w) { cb.onResult({}, `Worker ${workerId} not found`); return; }
this.callbacks.set(message.requestId, cb);
w.postMessage(message);
}
executeTask(workerId: string, taskType: string, params: Object, onProgress?: ProgressCallback): Promise<Object> {
return new Promise((resolve, reject) => {
const reqId = `task_${Date.now()}_${Math.random().toString(36).substring(7)}`;
const msg = new WorkerMessage(MessageType.EXECUTE, { taskType, params }, reqId);
this.sendMessage(workerId, msg, {
onResult: (r, e) => { e ? reject(new Error(e)) : resolve(r); },
onProgress
});
});
}
initWorker(workerId: string, initData: Object): Promise<Object> {
return new Promise((resolve, reject) => {
const msg = new WorkerMessage(MessageType.INIT, initData, `init_${workerId}_${Date.now()}`);
this.sendMessage(workerId, msg, { onResult: (r, e) => { e ? reject(new Error(e)) : resolve(r); } });
});
}
terminateWorker(workerId: string): void {
this.activeWorkers.get(workerId)?.terminate();
this.activeWorkers.delete(workerId);
}
terminateAll(): void { this.activeWorkers.forEach(w => w.terminate()); this.activeWorkers.clear(); }
private handleMessage(workerId: string, msg: WorkerMessage): void {
const cb = this.callbacks.get(msg.requestId);
if (!cb) return;
switch (msg.type) {
case MessageType.RESULT: cb.onResult(msg.payload); this.callbacks.delete(msg.requestId); break;
case MessageType.ERROR: cb.onResult({}, (msg.payload as { error: string }).error); this.callbacks.delete(msg.requestId); break;
case MessageType.PROGRESS: cb.onProgress?.((msg.payload as { percent: number; status: string }).percent, (msg.payload as { percent: number; status: string }).status); break;
}
}
getActiveCount(): number { return this.activeWorkers.size; }
}
4.4 完整使用示例
// ConcurrentDemo.ets — 并发编程完整演示页面
import { TaskPoolManager } from '../manager/TaskPoolManager';
import { WorkerManager } from '../manager/WorkerManager';
import { TaskConfig, TaskPriority, TaskType } from '../model/TaskModel';
import { processImageFilter, sortLargeDataset } from '../concurrent/concurrent_functions';
@Entry @Component
struct ConcurrentDemo {
@State logMessages: string[] = [];
@State taskCount: number = 0;
private taskPoolMgr: TaskPoolManager = TaskPoolManager.getInstance();
private workerMgr: WorkerManager = new WorkerManager();
private readonly WORKER_PATH = 'entry/ets/workers/LongRunningWorker.ts';
aboutToAppear(): void {
this.taskPoolMgr.createGroup('image_tasks');
this.taskPoolMgr.createGroup('compute_tasks');
this.workerMgr.createWorker(this.WORKER_PATH, 'long_compute');
this.appendLog('并发系统初始化完成');
}
aboutToDisappear(): void { this.workerMgr.terminateAll(); this.appendLog('并发系统已关闭'); }
async processBatchImages(paths: string[]): Promise<void> {
this.appendLog(`[TaskPool] 提交 ${paths.length} 个图像处理任务`);
const tasks = paths.map((path, idx) => ({
func: processImageFilter,
args: [`img_task_${idx}`, path, 'grayscale'] as [string, string, string],
config: new TaskConfig(`img_task_${idx}`, TaskType.IMAGE_PROCESS, TaskPriority.MEDIUM, 10000)
}));
const start = Date.now();
const results = await this.taskPoolMgr.submitBatch('image_tasks', tasks);
results.forEach((r, i) => {
r.success ? this.appendLog(`[完成] 任务${i}: ${r.duration}ms`) : this.appendLog(`[失败] ${r.errorMessage}`);
});
this.appendLog(`[TaskPool] 批次完成，耗时: ${Date.now() - start}ms`);
this.updateStats();
}
async sortLargeData(count: number): Promise<void> {
this.appendLog(`[TaskPool] 提交大数据排序，记录数: ${count}`);
const data: Array<number> = [];
for (let i = 0; i < count; i++) data.push(Math.random() * count);
const result = await this.taskPoolMgr.submitTask('compute_tasks', sortLargeDataset,
['sort_task_1', data, 'merge'], new TaskConfig('sort_task_1', TaskType.COMPUTATION, TaskPriority.HIGH, 30000));
if (result.success) {
const d = result.data as { sortedLength: number; isSorted: boolean };
this.appendLog(`[完成] 排序${d.sortedLength}条，验证${d.isSorted ? '通过' : '失败'}，耗时${result.duration}ms`);
} else { this.appendLog(`[失败] ${result.errorMessage}`); }
}
async runLongTask(taskType: string): Promise<void> {
this.appendLog(`[Worker] 提交长时任务: ${taskType}`);
try {
const result = await this.workerMgr.executeTask('long_compute', taskType,
{ frameCount: 100, recordCount: 50000, sources: 20 },
(percent, status) => this.appendLog(`[进度] ${percent}% - ${status}`));
this.appendLog(`[Worker] 任务完成: ${JSON.stringify(result)}`);
} catch (e) { this.appendLog(`[Worker] 错误: ${(e as Error).message}`); }
}
appendLog(msg: string): void {
const ts = new Date().toLocaleTimeString('zh-CN', { hour12: false });
this.logMessages.unshift(`[${ts}] ${msg}`);
if (this.logMessages.length > 50) this.logMessages.pop();
}
updateStats(): void {
const s = this.taskPoolMgr.getStats();
this.taskCount = s.totalSubmitted;
this.appendLog(`[统计] 提交:${s.totalSubmitted} 完成:${s.totalCompleted} 失败:${s.totalFailed}`);
}
build() {
Column() {
Text('ArkTS 并发编程演示').fontSize(24).fontWeight(FontWeight.Bold)
Text(`活动任务数: ${this.taskCount}`).fontSize(14).margin({ top: 4 })
List() {
ForEach(this.logMessages, (msg: string, idx: number) => {
ListItem() { Text(msg).fontSize(12).width('100%').fontColor(idx === 0 ? '#FF6600' : '#AAAAAA') }
})
}.height('50%').backgroundColor('#1A1A1A').padding(8)
Button('批量处理 5 张图片').width('90%').margin(4).onClick(() => this.processBatchImages(['/data/img1.jpg','/data/img2.jpg','/data/img3.jpg','/data/img4.jpg','/data/img5.jpg']))
Button('排序 20000 条数据').width('90%').margin(4).onClick(() => this.sortLargeData(20000))
Button('Worker 长任务').width('90%').margin(4).onClick(() => this.runLongTask('video_decode'))
}.padding(16)
}
}
五、深度技术原理
5.1 ArkTS 多线程模型
ArkTS 并发基于 Actor 并发模型：每个线程拥有独立内存空间，线程间通过消息传递通信，完全不存在锁竞争和共享数据竞争。好处是不需要考虑线程安全问题——没有共享变量就没有竞态条件。但跨线程传递数据时，数据必须被序列化（marshal）、通过 IPC 传输、在目标线程反序列化（unmarshal）为新对象，这一过程会产生性能开销。
主线程(UI渲染/事件分发) ──消息传递──▶ Worker线程A(独立堆内存A) / Worker线程B(独立堆内存B)
5.2 TaskPool 线程池调度算法
TaskPool 底层由系统管理固定数量后台线程，数量约为 CPU核心数 - 1（保留一核给 UI）。调度采用改进的**多级反馈队列（MLFQ）**策略：HIGH 优先级任务优先调度；每个任务执行固定时间片（约 10~20ms）后若未完成则放回队列尾部，防止单任务长期独占线程；低优先级任务等待过久时系统自动提升其优先级，避免饥饿。
5.3 序列化传输开销
| 数据类型 | 传输方式 | 10万条记录耗时 | 说明 |
|---|---|---|---|
| ArrayBuffer | 直接内存复制 | ~5ms | 最优，需包装为 Sendable |
| number[] | 数组整体复制 | ~30ms | Sendable 数组 |
| string (JSON) | JSON.stringify | ~120ms | 需手动序列化 |
| 普通 Object | structured clone | ~200ms | 复杂对象开销大 |
最佳实践：优先用 ArrayBuffer 传大数据；单个任务数据量不宜超过 1MB。
5.4 Worker 线程生命周期
Worker 与主线程长期绑定，直到显式调用 terminate() 才销毁。这意味着 Worker 实例可建立持久通信会话，支持多次往返消息。Worker 有独立堆内存，不会触发主线程 GC，适合长时重计算任务。但 Worker 数量需严格控制，同一 WorkerPath 实例不超过 CPU 核心数。
六、常见问题解答
Q1：TaskPool 和 Worker 如何选择？
A：任务执行时间小于 30 秒、调用频率高、需要优先级调度，选 TaskPool。任务超过 30 秒、需要独立执行上下文或多个 Worker 相互通信，选 Worker。两者可共存于同一应用。
Q2：@Concurrent 函数有什么限制？
A：必须是模块级顶层函数，不能是类方法。参数和返回值必须是 Sendable 类型。不支持访问未标记 Sendable 的全局状态或闭包。编译时 ArkTS 编译器进行类型检查，不符合规范的代码在编译阶段报错。
Q3：Sendable 和非 Sendable 对象的区别？
A：Sendable 对象跨线程传递时采用零拷贝引用传递，开销极低。非 Sendable 对象须经过结构化克隆完整序列化反序列化。对于包含数万条记录的大型数据结构，两者性能差异可达 10 倍。
Q4：任务超时后会自动取消吗？
A：TaskPool 不提供内置超时取消。超时控制需在调用层用 Promise.race 组合定时器，超时触发时调用 taskpool.cancel(task) 手动取消。取消是尽力而为的，若任务已执行完毕或处于不可中断计算阶段，取消可能无效。
Q5：Worker 线程中的错误如何处理？
A：Worker 通过 onerror 回调接收 Error 对象。Worker 内部应使用 try-catch 捕获计算错误，并通过 postMessage 发送 MessageType.ERROR 消息给主线程，由 WorkerManager 统一处理并触发 onResult 回调的错误分支。
Q6：多个 TaskPool 任务需要共享数据怎么办？
A：TaskPool 每个任务执行独立，不支持任务间直接内存共享。方案：每次调用时将共享数据作为参数传入（中小型数据）；使用 Worker 的持久状态（大量中间状态）；或使用 AppStorage（数据变化不频繁时）。
七、运行效果
八、扩展方向
ArkTS 并发模型仍在快速演进。第一，TaskPool 能力扩展：未来版本预计支持任务依赖图的声明式 DAG 调度，实现更复杂的任务编排。第二，共享容器（SharedContainer）：跨线程共享复杂数据结构的标准化方案，有望进一步降低大型数据集传递开销。第三，多 Worker 协作：Worker 间直接消息传递的引入，将显著提升大规模并行计算效率。第四，NPU 异构计算：TaskPool 有望支持将计算任务自动调度到 NPU 硬件核心，与 CPU 通用计算形成协同优化。
讨论HarmonyOS开发技术，专注于API与组件、DevEco Studio、测试、元服务和应用上架分发等。
更多推荐
5
0- 0
已为社区贡献31条内容
所有评论(0)