Dart 异步编程详解

Future 基础

Future 创建

Future<String> fetchUserData() {
  return Future.delayed(
    Duration(seconds: 2),
    () => 'User data',
  );
}

// 使用 async 关键字（同上函数的另一种写法）
Future<String> fetchUserDataAsync() async {
  await Future.delayed(Duration(seconds: 2));
  return 'User data';
}

Future 链式调用

Future<void> example() {
  return fetchUserData()
    .then((data) => processData(data))
    .then((result) => saveData(result))
    .catchError((error) => handleError(error));
}

async/await

基本用法

Future<void> loadUserData() async {
  try {
    String data = await fetchUserData();
    var result = await processData(data);
    await saveData(result);
  } catch (e) {
    print('Error: $e');
  }
}

并行执行

Future<void> loadMultipleData() async {
  try {
    final results = await Future.wait([
      fetchUserData(),
      fetchSettings(),
      fetchPreferences(),
    ]);
    print('All data loaded: $results');
  } catch (e) {
    print('Error: $e');
  }
}

Stream

Stream 创建

Stream<int> countStream(int max) async* {
  for (int i = 1; i <= max; i++) {
    await Future.delayed(Duration(seconds: 1));
    yield i;
  }
}

// StreamController
StreamController<String> controller = StreamController<String>();

Stream 监听

Future<void> listenToStream() async {
  // 方式一：await for
  await for (int value in countStream(5)) {
    print(value);
  }

  // 方式二：listen
  // 注意：单订阅 Stream 只能被监听一次，上面 await for 已经消费完一个实例，
  // 这里必须用新的 Stream 实例，否则抛 "Stream has already been listened to"
  countStream(5).listen(
    (data) => print(data),
    onError: (error) => print('Error: $error'),
    onDone: () => print('Stream completed'),
  );
}

完整示例

import 'dart:async';

class DataService {
  // 模拟网络请求
  Future<Map<String, dynamic>> fetchUserProfile(String userId) async {
    await Future.delayed(Duration(seconds: 2));
    return {
      'id': userId,
      'name': 'John Doe',
      'email': '[email protected]',
    };
  }
  
  Future<List<String>> fetchUserPosts(String userId) async {
    await Future.delayed(Duration(seconds: 1));
    return [
      'Post 1',
      'Post 2',
      'Post 3',
    ];
  }
  
  Stream<String> userActivityStream() async* {
    while (true) {
      await Future.delayed(Duration(seconds: 1));
      yield DateTime.now().toString();
    }
  }
}

class UserBloc {
  final DataService _dataService;
  final _activityController = StreamController<String>();
  StreamSubscription<String>? _activitySub;
  
  UserBloc(this._dataService);
  
  Stream<String> get activityStream => _activityController.stream;
  
  // 加载用户数据
  Future<void> loadUserData(String userId) async {
    try {
      // 并行加载数据
      final results = await Future.wait([
        _dataService.fetchUserProfile(userId),
        _dataService.fetchUserPosts(userId),
      ]);
      
      final profile = results[0] as Map<String, dynamic>;
      final posts = results[1] as List<String>;
      
      print('User Profile: $profile');
      print('User Posts: $posts');
    } catch (e) {
      print('Error loading user data: $e');
    }
  }
  
  // 监听用户活动
  void startActivityMonitoring() {
    _activitySub = _dataService.userActivityStream().listen(
      (activity) {
        _activityController.add('New activity: $activity');
      },
      onError: (error) {
        _activityController.addError('Error monitoring activity: $error');
      },
    );
  }
  
  Future<void> dispose() async {
    // 先取消对源 Stream 的订阅，再关闭 Controller：
    // 不 cancel 的话，userActivityStream 里的 while(true) 会让程序永远退不出
    await _activitySub?.cancel();
    await _activityController.close();
  }
}

void main() async {
  final dataService = DataService();
  final userBloc = UserBloc(dataService);
  
  // 加载用户数据
  await userBloc.loadUserData('123');
  
  // 监听用户活动
  userBloc.activityStream.listen(
    (activity) => print(activity),
    onError: (error) => print('Error: $error'),
  );
  
  userBloc.startActivityMonitoring();
  
  // 等待5秒后结束
  await Future.delayed(Duration(seconds: 5));
  await userBloc.dispose();
}

错误处理

Future 错误处理

Future<void> handleErrors() async {
  try {
    await riskyOperation();
  } catch (e) {
    print('Error caught: $e');
  } finally {
    print('Cleanup');
  }
}

Stream 错误处理

Stream<int> handleStreamErrors() async* {
  try {
    await for (var value in sourceStream()) {
      if (value < 0) {
        throw Exception('Negative value');
      }
      yield value;
    }
  } catch (e) {
    print('Error in stream: $e');
  }
}

最佳实践

1. 使用 async/await 而不是 then
2. 正确处理错误和异常
3. 及时关闭 StreamController
4. 避免过度使用异步操作
5. 合理使用并行执行

注意事项

1. 不要忽略返回的 Future
2. 注意 Stream 的订阅类型
3. 处理所有可能的错误
4. 避免死锁和循环等待
5. 合理管理异步资源

实战案例：串口连接的异步三件套

flutter_brightness 的 SerialService 把 Dart 异步的三件套用了个遍：

1. connect() 返回 bool 而不是 Future<bool>——注意，它是同步方法！打开串口、配 9600/8N1 这些操作本身不耗时（只是填结构体、调驱动），所以不需要 async。新手最容易犯的错就是”万物皆 async”，看到 IO 就加 async/await。判断标准很简单：这个操作会不会等？配串口参数不会等，读数据会等。

2. 读数据走 Stream。串口数据是源源不断来的，用 SerialPortReader 配 broadcast StreamController，UI 层 listen 就行。这里如果有人写成 while(true) { await read(); } 也不是不行，但 Stream 的好处是多个订阅者互不干扰，而且 cancel() 就能停。

3. 等外部设备的 Future 要有超时意识。串口设备可能没插、可能没反应——SerialPort.openReadWrite() 本身是同步快速失败的（打不开直接返回 false），这很好；但如果是”等待第一包数据”这种真正要等的 Future，就一定要包 timeout。所有等外部设备的 Future 都要有超时，这是铁律。没超时的等待，设备一拔，App 就卡死在那儿。

总结

Future 管”一次性的异步结果”，Stream 管”持续的事件流”，async/await 只是让写法变同步——三者分工明确，别混着用。写异步代码时多问一句”这个等待有没有超时、这个订阅有没有取消”，能避开一大半线上坑。