仓颉三方库实战：基于 async4cj 🚀

本文以本仓库 async4cj 为例，演示如何在 Cangjie（仓颉）项目中集成和使用第三方库，并通过集合处理与控制流原语构建可维护的业务流程。内容涵盖集成方式、常用场景与最佳实践，附完整示例代码。

快速了解 🧭

  • async4cj 将 async.js 风格的集合操作与控制流原语带到 Cangjie。
  • 当前实现为同步占位模型（single-thread），便于示例、教学与单元测试；后续可替换为真实异步调度。
  • 模块划分：📚 collections/、🔁 controlflow/、🧰 utils/、⚙️ internal/。

目录结构与特性能力说明 🗂️

  • 仓库结构（关键目录）

    • async4cj/cjpm.toml：包配置与编译器版本约束

    • async4cj/doc/：库文档与 API 导航

    • async4cj/src/

      ：仓颉源码（按能力模块拆分）

      • collections/：集合处理原语（映射、过滤、分组、去重等）
      • controlflow/：控制流编排（序列、并发限流、竞速、优先队列、重复执行等）
      • utils/：工具桥接（函数包装、同步函数转任务等）
      • internal/：内部类型与通用实现（比较器、键选择器、容器等）
    • async4cj/test/

      ：测试用例

      • HLT/：高层级场景测试（按能力拆分：controlflow_test.cj、priorityQueue_test.cj、race_test.cj）
      • LLT/：低层级单元测试（collections_test.cj、utils_test.cj）

  • 命名空间与能力图谱

    • collections：map、filter、groupKeys、uniq 等，适用于 ETL/数据管道与批量处理
    • controlflow：series（顺序执行）、parallelLimit（限流并发）、seq（函数管道）、race/raceBy（择优返回）、priorityQueueBy（按权重调度）、times/timesLimit（重复执行/限次）、nextTick/setImmediate（轻量调度占位）
    • utils：apply、asyncify 等，用于将同步函数桥接为零参任务或包装调用
    • internal：比较器、键选择器与内部辅助结构，供对外 API 复用

  • 能力边界（占位模型）

    • 单线程、确定性顺序，便于教学与测试；“并发/竞速”通过批次/择优模拟，非真实并发
    • 若涉及线程/IO，请替换底层调度或在真实运行时集成调度器；更多细节见“常见坑 ⚠️”一节

  • 适用场景速览

    • 批量变换与筛选：collections.*
    • 顺序/批次编排：series、parallelLimit
    • 竞速返回与优先级调度：race*、priorityQueueBy
    • 任务包装与轻量调度：apply/asyncify、nextTick/setImmediate

集成方式 📦

推荐两种方式之一：

  1. 子模块/源码方式（适合私有或试用阶段）

  • 在你的项目根目录添加子模块：git submodule add https://gitcode.com/cj-awaresome/async4cj.git third_party/async4cj

  • 将 third_party/async4cj/src 加入编译路径（依据你的构建脚本/工具设定）。

  • 源码内直接引入：

    import async4cj.collections.*
import async4cj.controlflow.*
import async4cj.utils.*
  1. 包管理方式（如你的 cjpm 已配置依赖）

  • 在

    cjpm.toml

    的

    [dependencies]

    下声明（示例为 path 依赖；实际以你的 cjpm 规范为准）：

    [dependencies]
async4cj = { path = "../third_party/async4cj" }

  • 然后在源码中按上文方式 import 即可。

构建与测试：

cjpm update
cjpm build
cjpm test

常用场景实战 🔨

1. 集合处理（ETL/数据管道）📚

过滤与映射：

import async4cj.collections.*

func transform(xs: Array<Int64>): Array<Int64> {
    let evens = filter<Int64>(xs, { x: Int64 => x % 2 == 0 })
    let doubled = map<Int64, Int64>(evens, { x: Int64 => x * 2 })
    return doubled
}

分组与去重：

import async4cj.collections.*

func uniqueKeys(xs: Array<Int64>): Array<Int64> {
    func key(x: Int64): Int64 { return x % 10 }
    func equals(a: Int64, b: Int64): Bool { return a == b }
    let keys = groupKeys<Int64, Int64>(xs, key, equals)
    return uniq<Int64>(keys, equals)
}

2. 控制流编排（任务序列/并发批次）🔁

顺序与分批执行：

import async4cj.controlflow.*

func runSeries(): Array<Int64> {
    let tasks = [{ => 1 }, { => 2 }, { => 3 }]
    return series<Int64>(tasks)
}

func runParallelBatches(): Array<Int64> {
    let tasks = [{ => 10 }, { => 20 }, { => 30 }, { => 40 }]
    return parallelLimit<Int64>(tasks, 2)
}

函数序列（管道/组合）：

import async4cj.controlflow.*

func pipeline(x0: Int64): Int64 {
    func add1(x: Int64): Int64 { return x + 1 }
    func times2(x: Int64): Int64 { return x * 2 }
    return seq<Int64>([add1, times2], x0)   // (x+1)*2
}

3. 竞速选择（择优返回）🏁

首个结果或按比较器最佳：

import async4cj.controlflow.*

func chooseFirst(): Int64 {
    return race<Int64>([{ => 5 }, { => 9 }, { => 1 }])
}

func chooseMax(): Int64 {
    func key(x: Int64): Int64 { return x }
    func less(a: Int64, b: Int64): Bool { return a < b }
    return raceBy<Int64, Int64>([{ => 5 }, { => 9 }, { => 1 }], key, less)
}

4. 优先队列（按权重处理）🎯

降序优先处理，结合并发批次：

import async4cj.controlflow.*

func processByPriority(items: Array<Int64>, prios: Array<Int64>): Array<Int64> {
    let order = Array<Int64>(items.size, { _ : Int64 => 0 })
    let idxBox = Array<Int64>(1, { _ : Int64 => 0 })
    func worker(x: Int64): Unit { order[idxBox[0]] = x; idxBox[0] = idxBox[0] + 1 }
    priorityQueueBy<Int64, Int64>(items, worker, 2, { x: Int64 => prios[(x / 10) - 1] }, { a: Int64, b: Int64 => a < b })
    return order
}

5. 同步函数桥接（任务包装）🧰

将同步函数包装为零参任务：

import async4cj.utils.*

func wrapExample(): Int64 {
    func f(): Int64 { return 7 }
    let t = asyncify<Int64>(f)
    return t()
}

项目结构与测试建议 🧪

  • 按能力划分测试文件：
    • HLT（高层级场景）：test/HLT/controlflow_test.cj、priorityQueue_test.cj、race_test.cj
    • LLT（低层级单元）：test/LLT/collections_test.cj、utils_test.cj
  • 使用 @Test 类 + @TestCase 方法，保证可读性与可维护性。

最佳实践 ✅

  • 纯函数优先：避免共享可变状态，提升可测试性。
  • 命名一致：按能力与模块命名（如 *_test.cj）。
  • 小步构建：先 series/seq 验证流程，再扩展 parallelLimit/priorityQueue。
  • 边界明确：limit/并发度与优先级策略要清晰约定。

常见坑 ⚠️

  • 当前实现为同步占位，不提供真实异步；若有线程/IO 需求需替换底层调度。
  • race/priorityQueue 的“并发”语义为顺序批次模拟，注意期望与实际实现对齐。

完整示例（main）📄

import async4cj.collections.*
import async4cj.controlflow.*
import async4cj.utils.*

main(): Int64 {
    let xs = [1, 2, 3, 4]
    let ys = map<Int64, Int64>(xs, { x: Int64 => x * x })

    let tasks = [{ => 1 }, { => 2 }, { => 3 }]
    let res = parallelLimit<Int64>(tasks, 2)

    func inc(x: Int64): Int64 { return x + 1 }
    let t = apply<Int64, Int64>(inc, 41)

    println("ys.size = ${ys.size}")
    println("res[0] = ${res[0]}")
    println("apply(41) => ${t()}")
    return 0
}

更多业务场景示例 💼

订单处理流水（校验→补全→计费→通知）

import async4cj.controlflow.*

func processOrder(x0: Int64): Int64 {
    func validate(x: Int64): Int64 { return x }              // 业务校验
    func enrich(x: Int64): Int64 { return x + 10 }           // 数据补全
    func charge(x: Int64): Int64 { return x * 2 }            // 计费计算
    func notify(x: Int64): Int64 { return x }                // 通知占位
    return seq<Int64>([validate, enrich, charge, notify], x0)
}

并发批次上传（限流上传队列）

import async4cj.controlflow.*

func uploadBatches(): Array<Int64> {
    // 以数字占位模拟文件上传任务
    let tasks = [{ => 101 }, { => 102 }, { => 103 }, { => 104 }]
    // 每批 2 个并行（占位实现顺序收集结果）
    return parallelLimit<Int64>(tasks, 2)
}

冗余服务竞速（择优返回）

import async4cj.controlflow.*

func queryWithFallback(): Int64 {
    // 模拟多个后端的冗余查询，返回首个结果
    return race<Int64>([{ => 200 }, { => 404 }, { => 503 }])
}

权重优先工单处理（按优先级调度）

import async4cj.controlflow.*

func scheduleTickets(items: Array<Int64>, prios: Array<Int64>): Array<Int64> {
    let order = Array<Int64>(items.size, { _ : Int64 => 0 })
    let idxBox = Array<Int64>(1, { _ : Int64 => 0 })
    func worker(x: Int64): Unit { order[idxBox[0]] = x; idxBox[0] = idxBox[0] + 1 }
    // 根据 prios 降序处理（并发度=2 批次执行）
    priorityQueueBy<Int64, Int64>(items, worker, 2, { x: Int64 => prios[(x / 10) - 1] }, { a: Int64, b: Int64 => a < b })
    return order
}

批量任务执行（序列/限次）

import async4cj.controlflow.*

func batchRun(): Int64 {
    let cBox = Array<Int64>(1, { _ : Int64 => 0 })
    times(5, { _ : Int64 => cBox[0] = cBox[0] + 1 })       // 执行 5 次
    timesLimit(7, 3, { _ : Int64 => cBox[0] = cBox[0] + 1 }) // 限次执行
    return cBox[0]
}

延迟与下一轮调度（轻量调度占位）

import async4cj.controlflow.*

func schedule(): Bool {
    let ranBox = Array<Bool>(1, { _ : Int64 => false })
    nextTick({ => ranBox[0] = true })     // 下一轮
    // 或者 setImmediate({ => ranBox[0] = true })
    return ranBox[0]
}

订单批量重试与告警（series + parallelLimit）

说明：批量处理订单，单个订单在占位模型下按最多 N 次重试（用 series 顺序尝试）；批量处理使用并发上限（用 parallelLimit）；最终仍失败的订单通过告警侧路发送（用 nextTick + series）。

import async4cj.collections.*
import async4cj.controlflow.*
import async4cj.utils.*

// 单个订单重试，占位：除最后一次外均失败；真实场景替换为接口调用
func retryProcessOrder(orderId: Int64, maxRetries: Int64): Bool {
    let attempts = Array<Bool>(maxRetries, { i: Int64 => i == (maxRetries - 1) })
    let attemptTasks = Array<() => Bool>(maxRetries, { i: Int64 => { => attempts[i] } })
    let results = series<Bool>(attemptTasks)
    // 汇总是否有成功
    let okBox = Array<Bool>(1, { _ : Int64 => false })
    times(results.size, { i: Int64 => if results[i] { okBox[0] = true } })
    return okBox[0]
}

// 批量处理，按并发上限执行每个订单的重试流程
func processOrdersWithRetry(orderIds: Array<Int64>, maxRetries: Int64, limit: Int64): Array<Bool> {
    let tasks = Array<() => Bool>(orderIds.size, { i: Int64 => { => retryProcessOrder(orderIds[i], maxRetries) } })
    return parallelLimit<Bool>(tasks, limit)
}

// 告警侧路：对失败订单生成并顺序执行告警任务，并放到下一轮调度
func sendAlerts(orderIds: Array<Int64>, results: Array<Bool>): Unit {
    let alertTasks = Array<() => Unit>(orderIds.size, { i: Int64 => {
        => if !results[i] { println("ALERT: order ${orderIds[i]} failed after retries") }
    }})
    nextTick({ => series<Unit>(alertTasks) })
}

// 演示总入口：最多重试 3 次，并发度 2，失败走告警侧路
func retryAndAlertDemo(): Array<Bool> {
    let orders = [101, 102, 103, 104]
    let results = processOrdersWithRetry(orders, 3, 2)
    sendAlerts(orders, results)
    return results
}

参考与扩展 🔗

  • 版本与编译器：见 async4cj/cjpm.toml（cjc-version = 1.0.3）
  • 许可证：MIT（见仓库根目录 LICENSE）
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

Rust 实战项目：TODO 管理器

avatar AtomGit开源社区
cover

使用rust复刻linux经典命令：wc（文本统计工具）

avatar AtomGit开源社区

如何用【rust】做一个命令行版的电子辞典

在语言学习中，「高频复习 + 个性化积累」是掌握单词的核心方法，但市面上多数单词 APP 存在广告冗余、功能繁杂、依赖网络等问题。本项目基于 Rust 标准库开发一款零依赖本地单词本，无需网络、无广告干扰、数据完全自主掌控，支持单词的增删查改、分类标签、复习测试等核心功能，兼顾实用性与学习价值。

avatar AtomGit开源社区