最近我给一个本地素材库加了“缩略图内容哈希”。目标很简单：导入一批图片后，在后台算出轻量 Hash，用来辅助重复素材判断。第一版我直接把 48 个文件全部扔进 TaskPool，页面看起来很流畅，但只要用户快速切换相册、重新筛选一次，问题马上出现：同一路径会被重复提交，已经不需要的任务还在继续跑，旧任务稍晚返回以后甚至会覆盖新一轮列表结果。

真正难的不是并发本身，而是任务身份、取消时机和结果归属。Demo ThumbHashGuardLab 的批次为 thumb_hash_091：48 个文件最终执行 36 个、取消 12 个、丢弃 3 个迟到结果，generation 为 5，结束后 Active Tasks=0。

一、TaskPool 只负责并发，业务还得自己定义“同一个任务是谁”

最开始我的代码每次进入可视区都 new 一个 taskpool.Task。如果同一张图因为列表复用、筛选刷新又进来一次，就会再提交一份完全相同的 Hash 计算。

系统能执行它们，但不知道这两个任务在业务上其实是同一个。

所以我先加一层 Registry，Key 直接使用素材路径：

import { taskpool } from '@kit.ArkTS'

@Concurrent
function hashThumbnail(path: string): string {
  let sum = 0

  for (let i = 0; i < 500000; i++) {
    sum = (sum + i * 97) & 0x7fffffff
  }

  return `${path}:${sum.toString(16)}`
}

private submitHash(path: string): void {
  if (this.taskRegistry.has(path)) {
    this.duplicateSubmitBlocked++
    return
  }

  const generation = this.latestGeneration
  const task = new taskpool.Task(
    hashThumbnail,
    path
  )

  this.taskRegistry.set(path, {
    task,
    generation
  })

  void this.executeHash(path, task, generation)
}

这里解决的是重复提交的业务定义：同一个 path 在当前 generation 里只允许一个活动 Task。若算法版本变化，Key 应升级为 path + algorithmVersion。

二、执行结果回来以前，要先问一句：它还属于当前页面吗

TaskPool 的 Promise 返回只是说明子任务算完了，不代表这个结果现在还应该进入 UI。

用户可能已经切换目录，或者重新发起了一批任务。旧任务如果比新任务晚返回，就会形成典型的“迟到结果覆盖新状态”。

我用 generation 做结果门禁：

private async executeHash(
  path: string,
  task: taskpool.Task,
  generation: number
): Promise<void> {
  try {
    const result =
      await taskpool.execute(task) as string

    if (generation !== this.latestGeneration) {
      this.lateResultsDropped++
      return
    }

    this.hashResultMap.set(path, result)
    this.executed++
  } catch (_) {
    // cancel 后进入这里不再写 UI
  } finally {
    this.taskRegistry.delete(path)
    this.activeTasks =
      this.taskRegistry.size
  }
}

这次测试里一共出现了 3 个迟到结果，因此最终 Late Results Dropped=3。

generation 只负责保证结果归属正确；如果结果还会写数据库，也要把 generation / batchId 带进持久化层。

三、页面已经不需要的任务，要主动 cancel，而不是等它自然结束

第二个问题来自相册切换。

用户从 48 个文件的目录切到另一个筛选条件后，其中 12 个 Hash 任务已经没有任何展示价值。如果让它们继续占 TaskPool，虽然最后结果会被 generation 丢弃，但 CPU 还是白跑了一遍。

我会在新的可见集合稳定后取消不再需要的任务：

private cancelStale(
  visiblePaths: Set<string>
): void {
  this.latestGeneration++

  for (const [path, record]
    of this.taskRegistry.entries()) {
    if (visiblePaths.has(path)) {
      continue
    }

    try {
      taskpool.cancel(record.task)
      this.cancelled++
    } catch (_) {
      // 任务可能已经完成，交给 finally 清理
    }

    this.taskRegistry.delete(path)
  }

  this.activeTasks =
    this.taskRegistry.size
}

本轮 48 个任务里，最终取消 12 个，真正执行 36 个。

generation 解决结果正确性，cancel 解决资源浪费。两层一起做，链路才完整。

四、不要在 @Concurrent 文件里顺手 import UI 状态

TaskPool 并发函数保持最小依赖，不引入 @Observed、AppStorage 等 UI 状态。Hash 核心只接收可序列化参数并返回纯结果。

页面状态、Registry、generation 全部留在宿主线程。

这次项目结构是：

entry/src/main/ets/
├─ pages/ThumbHashPage.ets
├─ concurrent/HashTaskRunner.ets
├─ concurrent/TaskRegistry.ets
└─ model/ThumbHashResult.ets

HashTaskRunner.ets 不 import 页面，也不碰 AppStorage。

如果 Hash 还要打开文件，open/close 也必须在并发函数内部成对处理。

五、批量任务不是越多越好，还要看当前业务是否真的需要

TaskPool 的价值是把 CPU 工作移出主线程，不意味着 48 个任务都必须长期有效。筛选变化后及时取消过期任务；长时间 I/O 也应优先用异步接口，避免占住工作线程。

六、调试页我只保留能判断并发是否收口的数据

HiLog 固定输出：

batch=thumb_hash_091 files=48

queued=48

cancelled=12

executed=36

lateResultsDropped=3 generation=5

activeTasks=0

State: RUNNING -> STABLE

运行结果里最关键的不是 184 ms，而是：

• Queued：48
• Executed：36
• Cancelled：12
• Late Results Dropped：3
• Active Tasks：0
• Latest Generation：5
• Last Task ID：62017
• Hash Cost：184 ms

Active Tasks=0 说明 Registry 没留下幽灵任务。

七、正式项目里还要补几个边界

正式项目还要注意：cancel 不是事务回滚；同一素材被多个页面消费时要做引用计数；高频任务通信不能直接变成 UI 刷新；应用退后台后是否继续计算要按产品场景决定。

八、这次我真正补上的，是 Task 的“归属感”

最后状态链路变成：

QUEUED → RUNNING → CANCELLING → STABLE

TaskPool 负责执行，Registry 负责去重，cancel 负责回收无效计算，generation 负责丢弃迟到结果。

并发任务真正稳定，不是因为“线程跑起来了”，而是每一个 Task 都知道自己为什么存在、什么时候失效，以及结果回来以后还能不能被当前页面接收。