Sunqy 文档库

主题

06 · Dart 异步并发与进阶语法 ​

【通用进阶】本文归入「02-Dart基础」类别，但阅读顺序排在 03~05 之后。 因为事件循环、Isolate、Stream 需要先理解 Flutter 的单线程渲染模型（03 章）才好消化。

开篇：前端概念对照 ​

前端概念Dart 对应差异
Promise<T>Future<T>几乎一样
async / await同几乎一样
事件循环（Event Loop）同，但有微任务队列JS 也有微任务，概念一致
Promise.allFuture.wait
Promise.race无内置，需手写
Web WorkerIsolate⚠️ 机制差异大（不共享内存）
RxJS ObservableStream<T>概念接近
async function*同（异步生成器）一样
for await...ofawait for一样
TS extends 混入mixinDart 原生支持
TS 声明合并 / prototype 扩展extensionDart 原生支持
泛型约束 T extends X<T extends X>一样

一、单线程模型与事件循环 ​

为什么 Flutter 是单线程 ​

Flutter 的 UI 渲染必须在一个线程上完成（保证渲染一致性）。Dart 采用单线程 + 事件循环模型：

┌─────────────────────────────────────────┐
│             Dart Isolate                 │
│                                          │
│   微任务队列 (Microtask Queue)  ← 高优先级 │
│            ↓                             │
│   事件队列 (Event Queue)       ← 普通     │
│            ↓                             │
│       Event Loop 不断取出执行              │
└─────────────────────────────────────────┘

执行顺序规则 ​

  1. 先执行同步代码
  2. 清空微任务队列
  3. 从事件队列取一个任务执行
  4. 回到第 2 步（每执行一个事件任务，就清空一次微任务队列）

与 JS 的对比 ​

几乎完全一样。如果你熟悉 JS 的 Promise.then（微任务）vs setTimeout（宏任务），你已经懂了。

JSDart
Promise.then / queueMicrotask / MutationObserverFuture.microtask / scheduleMicrotask
setTimeout / setInterval / I/OFuture.delayed / Timer / I/O / 手势事件
async 函数里的 await同（await 之后的代码变成微任务）

代码验证 ​

dart
void main() {
  print('1 同步开始');

  Future(() => print('4 事件任务 1'));
  Future(() => print('5 事件任务 2'));

  scheduleMicrotask(() => print('3 微任务'));

  Future.delayed(Duration.zero, () => print('6 delayed'));

  print('2 同步结束');
}

输出顺序：1 → 2 → 3 → 4 → 5 → 6

解释：

  1. 同步代码先执行完（1、2）
  2. 清空微任务队列（3）
  3. 依次执行事件任务（4、5、6）

⚠️ 关键陷阱：微任务饥饿 ​

不要在微任务里无限产生微任务，否则事件队列永远得不到执行，UI 会卡死。

dart
void badLoop() {
  scheduleMicrotask(() {
    // 又产生一个微任务 → 事件队列永远排不上 → 界面冻结
    badLoop();
  });
}

实际影响：微任务里只放短小的、必须原子完成的逻辑。

与 Flutter 渲染的关系 ​

Flutter 的渲染帧也是通过事件队列调度的。所以：

任何长时间占用主线程的操作都会导致掉帧（jank）。

dart
// ❌ 会卡顿（占用主线程 3 秒）
void onPressed() {
  final result = heavyComputation();   // 同步密集计算
}

// ✅ 密集计算应该丢进 Isolate（见第三节）

前端类比：等同于 JS 里跑一个大循环阻塞渲染。

练习 6.1：验证执行顺序 ​

dart
// 粘到 lib/main_playground.dart 的 main() 或按钮回调
void orderDemo() {
  debugPrint('A 同步');
  Future(() => debugPrint('D 事件1'));
  Future.microtask(() => debugPrint('C 微任务'));
  Future(() => debugPrint('E 事件2'));
  scheduleMicrotask(() => debugPrint('B 微任务2'));
  debugPrint('A2 同步结束');
}

先自己预测顺序，再运行验证。

自检 ​

  • [ ] 理解「同步 → 清空微任务 → 执行一个事件任务 → 循环」
  • [ ] 知道 await 之后的代码会进入微任务队列
  • [ ] 知道长时间同步计算会导致掉帧

二、Future 全家桶 ​

创建 Future ​

dart
Future.value(42)                                  // 立即完成
Future.error(Exception('出错了'))                  // 立即失败
Future.delayed(Duration(seconds: 1), () => 42)     // 延迟
Future.microtask(() => 42)                        // 微任务
Future.sync(() => 42)                             // 同步执行

async 函数 ​

dart
Future<int> fetchCount() async {
  await Future.delayed(const Duration(seconds: 1));
  return 42;    // 自动包装成 Future.value(42)
}

⚠️ async 函数里抛出的异常会自动变成 rejected Future：

dart
Future<int> bad() async {
  throw Exception('boom');      // 等价于 return Future.error(...)
}

并发：Future.wait ​

dart
// ❌ 串行：总共 3 秒
final a = await fetch(1);   // 1 秒
final b = await fetch(2);   // 1 秒
final c = await fetch(3);   // 1 秒

// ✅ 并发：总共 1 秒
final results = await Future.wait([fetch(1), fetch(2), fetch(3)]);
// results 是 List<int>

前端对照：Promise.all。

⚠️ Future.wait 的失败行为：默认快速失败（任一失败则整体失败）。

dart
// 想让部分失败不影响整体
final results = await Future.wait(
  tasks,
  eagerError: false,                    // 不快速失败
);

但即便如此，Future.wait 遇到 error 仍会抛出。要收集所有结果（含错误），用 wait 配合 catchError：

dart
final results = await Future.wait(
  tasks.map((t) => t.catchError((e) => null)),   // 把错误转成 null
);

超时：timeout ​

dart
try {
  final data = await fetchData().timeout(
    const Duration(seconds: 5),
    onTimeout: () => '默认数据',      // 超时后的兜底值
  );
} on TimeoutException {
  // 如果不给 onTimeout，会抛 TimeoutException
}

错误处理 ​

dart
// 方式一：try/catch（推荐）
try {
  final data = await fetchData();
} on DioException catch (e) {
  // 特定类型
} catch (e, stack) {
  // 其他
} finally {
  // 清理
}

// 方式二：catchError 链式
await fetchData().catchError((e) => defaultValue);

⚠️ 陷阱：忘记 await 导致错误被吞 ​

dart
// ❌ 没有 await，异常变成未处理的 Future 错误
fetchData();

// ✅
await fetchData();
// 或
unawaited(fetchData());    // 明确表示「故意不等待」

本项目用 dart:async 的 unawaited 表达「故意不等待」的意图，例如 lib/main.dart:80：

dart
Future.delayed(const Duration(seconds: 5), FlutterNativeSplash.remove);

以及 lib/main.dart:251-255：

dart
unawaited(
  ref.read(i18nSyncServiceProvider)
      .sync(resolveEffectiveLocale(ref.read(localeNotifierProvider))),
);

unawaited 的价值：明确告诉读者和 linter「我不是忘了 await，我是故意不等」。

练习 6.2：并发 vs 串行 ​

dart
Future<int> slow(int n) async {
  await Future.delayed(const Duration(milliseconds: 500));
  return n * 2;
}

Future<void> compareDemo() async {
  // 1. 测串行耗时
  final sw1 = Stopwatch()..start();
  final a = await slow(1);
  final b = await slow(2);
  final c = await slow(3);
  sw1.stop();
  debugPrint('串行: ${sw1.elapsedMilliseconds}ms, 结果 ${[a, b, c]}');

  // 2. 测并发耗时
  final sw2 = Stopwatch()..start();
  final results = await Future.wait([slow(1), slow(2), slow(3)]);
  sw2.stop();
  debugPrint('并发: ${sw2.elapsedMilliseconds}ms, 结果 $results');
}

观察：串行 ~1500ms，并发 ~500ms。

自检 ​

  • [ ] 会用 Future.wait 做并发
  • [ ] 知道 Future.wait 默认快速失败
  • [ ] 会用 timeout 做超时兜底
  • [ ] 知道故意不等待时用 unawaited

三、Isolate：Dart 的唯一真并发 ​

为什么需要它 ​

单线程 + 密集计算 = 掉帧。Isolate 把计算丢到另一个线程去跑。

前端类比与关键差异 ​

Web WorkerDart Isolate
多线程✅✅
共享内存❌（但可 SharedArrayBuffer）❌ 完全不共享
通信方式postMessage（结构化克隆）SendPort / ReceivePort（消息传递）
传递大对象拷贝（或 Transferable）拷贝（大对象有成本）

核心差异：Isolate 不共享任何内存。它们各自有独立的堆和事件循环。

这意味着：

  • ✅ 没有数据竞争，不需要锁
  • ❌ 传数据要序列化拷贝，大对象有成本

两种用法 ​

用法一：compute（简单场景，推荐） ​

Flutter 封装的便捷 API，适合一次性的、有输入有输出的计算：

dart
// 顶层函数或静态方法（不能是闭包/实例方法）
int heavySum(int n) {
  int sum = 0;
  for (var i = 0; i < n; i++) {
    sum += i;
  }
  return sum;
}

// 调用
final result = await compute(heavySum, 100000000);

⚠️ compute 的限制（新手常踩）：

限制原因
必须是顶层函数或静态方法Isolate 间不共享闭包捕获的上下文
参数和返回值必须可序列化需要跨 Isolate 传输
不能捕获 context、回调等同上
dart
// ❌ 错误：匿名函数不行
await compute((n) => n * 2, 10);

// ✅ 正确：顶层/静态函数
int doubleIt(int n) => n * 2;
await compute(doubleIt, 10);

用法二：Isolate.spawn（长驻、多轮通信） ​

适合需要反复通信的场景（如持续的数据处理管道）：

dart
void isolateEntry(SendPort sendPort) {
  final receivePort = ReceivePort();
  sendPort.send(receivePort.sendPort);   // 把自己的端口发回去

  receivePort.listen((message) {
    if (message is int) {
      sendPort.send(heavySum(message));   // 处理并发回结果
    }
  });
}

// 主 Isolate
final receivePort = ReceivePort();
await Isolate.spawn(isolateEntry, receivePort.sendPort);

final sendPort = await receivePort.first as SendPort;

// 之后可以反复通信
sendPort.send(1000);

日常开发中 compute 够用，Isolate.spawn 只在复杂场景用。

什么时候该用 Isolate ​

场景是否用 Isolate
网络请求❌ 不需要（本身就是异步 I/O，不阻塞线程）
文件读写❌ 不需要（异步 I/O）
JSON 解析（大 JSON）⚠️ 数据量大时考虑
图片压缩/处理✅ 需要
复杂数学计算✅ 需要
大量数据的排序/过滤⚠️ 看数据量（万级以上考虑）

💡 经验法则：如果这个操作让你看到掉帧（< 16ms 内完不成），就该考虑 Isolate。

Flutter 一帧的预算是 16.67ms（60fps）。超过就会掉帧。

本项目用了 Isolate 吗 ​

没有（至少不在业务代码里）。本项目的数据处理量不足以需要它，且：

  • 网络请求本身异步，不阻塞
  • 分页列表每次只加载 10~20 条，解析成本极低

所以你大概率不会在本项目写 Isolate——但要知道它的存在和适用边界，避免把「卡」误判成需要多线程。

自检 ​

  • [ ] 知道 Isolate 不共享内存，靠消息通信
  • [ ] 知道 compute 的函数必须是顶层/静态
  • [ ] 知道网络请求不需要 Isolate（本身就是异步 I/O）
  • [ ] 知道判断标准：操作是否超过 16ms

四、Stream：异步数据流 ​

前端对照 ​

概念JSDart
异步序列RxJS Observable / AsyncIterableStream<T>
订阅subscribe()listen()
取消订阅unsubscribe()cancel()（返回的是 StreamSubscription）
变换map / filter / debouncemap / where / debounce（需 rxdart）
生成器async function*async*
消费for awaitawait for

创建 Stream ​

dart
// 方式一：async* 生成器（最简洁）
Stream<int> countStream(int max) async* {
  for (var i = 1; i <= max; i++) {
    await Future.delayed(const Duration(seconds: 1));
    yield i;      // 产出一个值
  }
}

// 方式二：StreamController（需要外部控制）
final controller = StreamController<int>();
controller.add(1);          // 从外部推数据
controller.addError('err');
controller.close();          // ⚠️ 必须关闭

⚠️ StreamController 必须 close()，否则泄漏。通常在 dispose 里做（呼应 05 章）。

消费 Stream ​

dart
// 方式一：await for（适合顺序处理）
await for (final value in countStream(5)) {
  print(value);
}

// 方式二：listen（适合响应式处理）
final subscription = countStream(5).listen(
  (value) => print('收到 $value'),
  onError: (e) => print('错误 $e'),
  onDone: () => print('完成'),
);

// 记得取消
subscription.cancel();

Stream 的两种类型 ​

类型特点用途前端类比
单订阅（默认）只能 listen 一次一次性数据流（文件读取、网络下载）Promise-like
广播（asBroadcastStream()）可多个订阅者事件总线、状态变更通知EventEmitter
dart
// 转成广播
final broadcast = controller.stream.asBroadcastStream();

// 或创建时指定
final controller = StreamController<int>.broadcast();

⚠️ 对单订阅 Stream listen 两次会报错：Bad state: Stream has already been listened to.

常用变换 ​

dart
stream
  .where((e) => e > 0)              // 过滤（≈ filter）
  .map((e) => e * 2)                 // 变换
  .take(5)                           // 只取前 5 个
  .distinct()                        // 去重（相邻相同）
  .debounceTime(...)                 // 防抖（需 rxdart 包）
  .transform(...)                    // 自定义转换

StreamBuilder：在 UI 里用 Stream ​

dart
StreamBuilder<int>(
  stream: countStream(10),
  initialData: 0,
  builder: (context, snapshot) {
    if (snapshot.hasError) return Text('错误 ${snapshot.error}');
    switch (snapshot.connectionState) {
      case ConnectionState.waiting:
        return const CircularProgressIndicator();
      case ConnectionState.active:
        return Text('${snapshot.data}');
      case ConnectionState.done:
        return const Text('结束');
      default:
        return const SizedBox.shrink();
    }
  },
)

前端类比：RxJS 的 observable | async 管道。

⚠️ 本项目几乎不用 StreamBuilder —— 因为状态管理统一走 Riverpod 的 AsyncValue（见 08 章），语义更统一、更好测试。

但 Stream 本身在底层被广泛使用：

  • 网络库的响应流
  • AuthRefreshListenable（路由刷新监听，lib/routing/auth_refresh_listenable.dart）
  • 各种平台插件的事件通道

FutureBuilder：一次性异步的 UI ​

dart
FutureBuilder<String>(
  future: fetchData(),      // ⚠️ 注意：不要每次 build 都新建 future
  builder: (context, snapshot) {
    if (snapshot.connectionState == ConnectionState.waiting) {
      return const CircularProgressIndicator();
    }
    if (snapshot.hasError) return Text('${snapshot.error}');
    return Text('${snapshot.data}');
  },
)

⚠️ 经典 bug：future: fetchData() 写在 build 里，每次重建都会重新发起请求。

正确做法：把 future 存到 State 里。

dart
late final Future<String> _future;

@override
void initState() {
  super.initState();
  _future = fetchData();     // 只创建一次
}

// build 里用 future: _future

本项目不用 FutureBuilder——统一用 Riverpod 的 AsyncValue + LoadingStateHandler（见 08 章），它自动处理了「只请求一次」的问题。

练习 6.3：Stream 计时器 ​

dart
Stream<int> ticker(int seconds) async* {
  for (var i = seconds; i >= 0; i--) {
    await Future.delayed(const Duration(seconds: 1));
    yield i;
  }
}

class StreamDemo extends StatelessWidget {
  const StreamDemo({super.key});

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('Stream 倒计时')),
      body: Center(
        child: StreamBuilder<int>(
          stream: ticker(10),
          initialData: 10,
          builder: (context, snapshot) {
            if (snapshot.connectionState == ConnectionState.done) {
              return const Text('倒计时结束', style: TextStyle(fontSize: 32));
            }
            return Text('${snapshot.data}', style: const TextStyle(fontSize: 64));
          },
        ),
      ),
    );
  }
}

观察：每秒递减。connectionState 从 waiting → active → done。

自检 ​

  • [ ] 会用 async* + yield 创建 Stream
  • [ ] 知道 StreamController 必须 close()
  • [ ] 知道单订阅 Stream 不能 listen 两次
  • [ ] 知道本项目 UI 层不用 StreamBuilder/FutureBuilder，统一走 Riverpod

五、进阶语法：extension / mixin / 泛型 ​

5.1 extension：给已有类型加方法 ​

前端类比：TS 的声明合并 / JS 的 prototype 扩展，但 Dart 是类型安全的。

dart
extension StringX on String {
  bool get isBlank => trim().isEmpty;
  String capitalize() => isEmpty ? this : '${this[0].toUpperCase()}${substring(1)}';
}

// 使用
'  '.isBlank;              // true
'hello'.capitalize();      // 'Hello'

⚠️ extension 只在显式 import 了该扩展的文件里生效。

项目实例：本项目的两个核心 extension ​

① AppTokensContextExtension —— lib/he_components/src/themes/app_tokens.dart:544-545：

dart
extension AppTokensContextExtension on BuildContext {
  AppTokens get appTokens => AppTokens.of(this);
}

这就是 context.appTokens 能用的原因。

② AuthStateX —— lib/ui/auth/models/auth_state.dart:145-174：

dart
extension AuthStateX on AuthState {
  bool get isAuthenticated =>
      this is LoginAuthenticated || this is CompleteLoginAuthenticated;

  List<String>? get permissions { ... }

  bool hasPermission(String code) {
    final perms = permissions;
    if (perms == null || perms.isEmpty) return false;
    return perms.any((p) => p == '*:*:*' || p == code);
  }
}

为什么用 extension 而不是直接写进类：因为 AuthState 是 freezed 生成的类，源码不能直接改（改了会被代码生成覆盖）。extension 是给生成类「外挂」方法的最佳手段。

这个技巧在本项目非常常见，你会反复遇到。

③ PermissionStatusExt —— lib/core/permission/permission_status.dart:19-25：

dart
extension PermissionStatusExt on PermissionStatus {
  bool get isGranted => this == PermissionStatus.granted;
  bool get isPermanentlyDenied => this == PermissionStatus.permanentlyDenied;
  bool get isDenied => this == PermissionStatus.denied || isPermanentlyDenied;
}

5.2 mixin：横向复用代码 ​

前端类比：TS 的 mixin 模式，但 Dart 原生支持。

dart
mixin LoggerMixin {
  void log(String msg) => debugPrint('[LOG] $msg');
}

class MyViewModel with LoggerMixin {
  void doSomething() {
    log('做了件事');    // 直接用
  }
}

项目实例：分页 mixin ​

lib/ui/core/notifiers/paged_notifier_mixin.dart:28-29 是本项目 mixin 的典范：

dart
mixin PagedNotifierMixin<TOrder, TDto>
    on AutoDisposeAsyncNotifier<PagedState<TOrder>> {
  ...
}

拆解：

  • mixin PagedNotifierMixin<TOrder, TDto>：声明一个泛型 mixin
  • on AutoDisposeAsyncNotifier<PagedState<TOrder>>：约束——只能混入到这个类型的类上

on 约束的意义：mixin 内部可以直接调用宿主类（AutoDisposeAsyncNotifier）的成员，如 state：

dart
state = const AsyncValue.loading();      // ← state 来自宿主类
state = await AsyncValue.guard(() => _fetchPage(_page));

使用方式：

dart
@riverpod
class PlantSelectListViewModel extends _$PlantSelectListViewModel
    with PagedNotifierMixin<PlantDto, PlantDto> {
  @override
  Future<PagedState<PlantDto>> build() => initialLoad();

  @override
  Future<Result<PageDTO<PlantDto>>> fetchPage(int page, {...}) async { ... }

  @override
  PlantDto mapDto(PlantDto dto) => dto;
}

子类只需实现两个抽象方法 fetchPage 和 mapDto，分页的全部逻辑（搜索、刷新、加载更多、判终）都由 mixin 提供。

前端类比：类似 React 的自定义 Hook 抽取公共逻辑，或 Vue 的 composable。

mixin vs 继承 vs 接口：

用法语义
extends单继承「是一种」
implements实现接口「符合契约」
with（mixin）混入复用「具备某能力」
on约束 mixin 宿主「只能混入这类」

5.3 泛型与泛型边界 ​

dart
// 基础泛型
class Box<T> {
  final T value;
  Box(this.value);
}

// 泛型边界：T 必须是某类型的子类
class NumberBox<T extends num> {
  final T value;
  NumberBox(this.value);
  double get doubleValue => value.toDouble();   // ✅ 因为有边界，能调 num 的方法
}

// 泛型函数
T firstOf<T>(List<T> list) => list.first;

前端对照：和 TS 的 class Box<T>、<T extends num> 完全一致。

项目实例：本项目的高频泛型 ​

dart
// 分页状态泛型（lib/ui/core/models/paged_state.dart）
PagedState<T>

// 分页 mixin 双泛型：UI 模型 + DTO
mixin PagedNotifierMixin<TOrder, TDto> on AutoDisposeAsyncNotifier<PagedState<TOrder>>

// 结果类型泛型
Result<T>

5.4 typedef：函数类型别名 ​

dart
typedef ItemBuilder<T> = Widget Function(BuildContext context, T item, int index);
typedef JsonMap = Map<String, dynamic>;

// 使用
void register(ItemBuilder<String> builder) { ... }

前端对照：TS 的 type ItemBuilder<T> = (...) => Widget。

5.5 late final：延迟一次的常量 ​

dart
late final String _computed;    // 第一次访问时才计算

String get computed => _computed;   // 惰性求值

前端类比：类似带缓存的 getter（@cached）。

自检 ​

  • [ ] 知道 extension 是给（尤其是生成的）类外挂方法的手段
  • [ ] 知道 AuthState.hasPermission 实际是 extension 提供的
  • [ ] 理解 mixin 的 on 约束及其意义
  • [ ] 能读懂 PagedNotifierMixin<TOrder, TDto>

六、本章速查 ​

异步 ​

需求写法
并发执行Future.wait([a, b, c])
超时兜底.timeout(d, onTimeout: () => fallback)
故意不等待unawaited(future)
微任务scheduleMicrotask / Future.microtask
延迟Future.delayed(duration, () => v)
密集计算compute(顶层函数, 参数)

Stream ​

需求写法
创建（生成器）Stream<T> f() async* { yield v; }
创建（控制器）StreamController<T>()（记得 close()）
顺序消费await for (final v in stream)
响应式消费stream.listen(...)（记得 cancel()）
多订阅.asBroadcastStream()
过滤变换.where() / .map() / .take()

语法 ​

需求写法
给类加方法extension X on Type { ... }
横向复用mixin M on Host { ... } + with M
泛型约束<T extends Base>
函数别名typedef F = R Function(A)
惰性常量late final T x;

下一步 ​

通用基础部分（01~06）到此结束。接下来进入本项目实战：

→ 07 · 项目架构总览