Dart是单线程语言，通过事件循环（Event Loop）处理异步操作。理解事件循环是掌握Dart异步编程的基石。
// Dart运行时结构：
// ┌─────────────────────────────────┐
// │ Main Isolate │
// │ ┌──────────┐ ┌────────────┐ │
// │ │ Microtask │ │ Event Queue│ │
// │ │ Queue │ │ │ │
// │ └────┬─────┘ └─────┬──────┘ │
// │ │ │ │
// │ ┌────▼──────────────▼──────┐ │
// │ │ Event Loop │ │
// │ │ 1. 先清空Microtask │ │
// │ │ 2. 取一个Event执行 │ │
// │ │ 3. 回到步骤1 │ │
// │ └─────────────────────────┘ │
// └─────────────────────────────────┘// ✅ 验证通过：Dart 3.4
// 文件：lib/async_basics.dart
/// 1. 创建Future
void creatingFutures() {
// 方式1：Future构造函数
Future<String> fetchName = Future(() {
return 'Flutter';
});
// 方式2：Future.value — 立即完成
Future<int> immediate = Future.value(42);
// 方式3：Future.error — 立即失败
Future<Never> failed = Future.error(Exception('出错了'));
// 方式4：Future.delayed — 延迟执行
Future<String> delayed = Future.delayed(
const Duration(seconds: 2),
() => '延迟2秒的结果',
);
// 方式5：async函数自动包装
Future<String> getData() async {
await Future.delayed(const Duration(seconds: 1));
return '异步数据';
}
// 方式6：Completer — 精确控制完成时机
// import 'dart:async';
// var completer = Completer<String>();
// completer.complete('手动完成');
// completer.future.then(print);
}
/// 2. 消费Future
void consumingFutures() async {
// 方式1：then/catchError/whenComplete
fetchUserData()
.then((data) => print('数据: $data'))
.catchError((error) => print('错误: $error'))
.whenComplete(() => print('完成（无论成功失败）'));
// 方式2：async/await（推荐！）
try {
var data = await fetchUserData();
print('数据: $data');
} catch (error) {
print('错误: $error');
} finally {
print('完成');
}
}
/// 3. Future组合
void combiningFutures() async {
// Future.wait — 并行执行，等待全部完成
var results = await Future.wait([
fetchUserData(),
fetchConfig(),
fetchPreferences(),
]);
print('全部完成: $results');
// Future.wait 带错误处理
try {
var all = await Future.wait([
Future.delayed(const Duration(seconds: 1), () => 'A'),
Future.delayed(const Duration(seconds: 2), () => 'B'),
Future.delayed(const Duration(seconds: 3), () => 'C'),
]);
print('并行结果: $all'); // ['A', 'B', 'C']
} catch (e) {
print('其中一个失败: $e');
}
// Future.any — 谁先完成用谁
var fastest = await Future.any([
Future.delayed(const Duration(seconds: 3), () => '慢服务'),
Future.delayed(const Duration(seconds: 1), () => '快服务'),
]);
print('最快的: $fastest'); // 快服务
// 顺序执行
var step1 = await stepOne();
var step2 = await stepTwo(step1);
var step3 = await stepThree(step2);
print('流水线结果: $step3');
}
/// 4. Future超时
void timeoutDemo() async {
try {
var result = await fetchUserData()
.timeout(const Duration(seconds: 5));
print('结果: $result');
} on TimeoutException {
print('请求超时！');
}
// 或者提供超时默认值
var result = await fetchUserData()
.timeout(
const Duration(seconds: 5),
onTimeout: () => '默认数据',
);
print(result);
}
// 模拟API
Future<String> fetchUserData() async {
await Future.delayed(const Duration(seconds: 1));
return '{"name": "Alice", "age": 25}';
}
Future<String> fetchConfig() async {
await Future.delayed(const Duration(milliseconds: 500));
return '{"theme": "dark"}';
}
Future<String> fetchPreferences() async {
await Future.delayed(const Duration(milliseconds: 800));
return '{"lang": "zh"}';
}
Future<String> stepOne() async => '步骤1完成';
Future<String> stepTwo(String input) async => '$input → 步骤2完成';
Future<String> stepThree(String input) async => '$input → 步骤3完成';如果说Future是"一次性的异步值"，那么Stream就是"异步的事件序列"——就像管道中持续流出的水。
// ✅ 验证通过
// 文件：lib/streams.dart
/// 1. 创建Stream
void creatingStreams() {
// 方式1：Stream.fromIterable
var stream1 = Stream.fromIterable([1, 2, 3, 4, 5]);
// 方式2：Stream.periodic
var stream2 = Stream.periodic(
const Duration(seconds: 1),
(count) => 'Tick $count',
);
// 方式3：async* 生成器
Stream<int> countStream(int max) async* {
for (var i = 1; i <= max; i++) {
await Future.delayed(const Duration(seconds: 1));
yield i;
}
}
// 方式4：StreamController
// var controller = StreamController<String>();
// controller.add('事件1');
// controller.add('事件2');
// controller.close();
// 方式5：Stream.fromFuture
var stream3 = Stream.fromFuture(fetchData());
}
/// 2. 消费Stream
void consumingStreams() async {
// 方式1：await for
await for (var value in countStream(5)) {
print('收到: $value');
}
// 方式2：listen
var subscription = countStream(3).listen(
(data) => print('数据: $data'),
onError: (error) => print('错误: $error'),
onDone: () => print('流结束'),
cancelOnError: false,
);
// 可以暂停和恢复
subscription.pause();
subscription.resume();
subscription.cancel(); // 取消订阅
}
/// 3. Stream变换
void streamTransformations() async {
var stream = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
// map — 变换每个元素
var doubled = stream.map((n) => n * 2);
// where — 过滤
var evens = stream.where((n) => n % 2 == 0);
// expand — 一对多
var expanded = stream.expand((n) => [n, n * 10]);
// take/skip
var first3 = stream.take(3);
var skip5 = stream.skip(5);
// distinct — 去重
var unique = Stream.fromIterable([1, 1, 2, 2, 3]).distinct();
// reduce/fold
var sum = await stream.reduce((a, b) => a + b);
var product = await stream.fold(1, (a, b) => a * b);
// asyncMap — 异步变换
var asyncMapped = stream.asyncMap((n) async {
await Future.delayed(const Duration(milliseconds: 100));
return n * n;
});
// debounce — 防抖
// 需要rxdart包：stream.debounce(Duration(milliseconds: 300))
// 组合变换
var result = stream
.where((n) => n % 2 == 0)
.map((n) => n * n)
.take(3);
print(await result.toList()); // [4, 16, 36]
}
/// 4. StreamController
class EventBus {
final _controller = StreamController<String>.broadcast();
Stream<String> get events => _controller.stream;
void emit(String event) {
if (!_controller.isClosed) {
_controller.add(event);
}
}
void dispose() {
_controller.close();
}
}
/// 5. 实时数据源
class StockTicker {
final String symbol;
final _controller = StreamController<double>.broadcast();
Timer? _timer;
StockTicker(this.symbol);
Stream<double> get prices => _controller.stream;
void start() {
var price = 100.0;
_timer = Timer.periodic(const Duration(seconds: 1), (_) {
// 模拟价格波动
price += (DateTime.now().millisecond % 100 - 50) * 0.1;
price = price.clamp(50.0, 200.0);
_controller.add(price);
});
}
void stop() {
_timer?.cancel();
_controller.close();
}
}
Future<String> fetchData() async {
await Future.delayed(const Duration(seconds: 1));
return '数据';
}
Stream<int> countStream(int max) async* {
for (var i = 1; i <= max; i++) {
await Future.delayed(const Duration(seconds: 1));
yield i;
}
}// ✅ 验证通过
// 文件：lib/isolates.dart
// import 'dart:isolate';
/// Isolate是Dart的并发单元——真正的并行执行
/// 每个Isolate有自己的内存堆，不共享内存
/// 通过SendPort/ReceivePort通信
// 简单Isolate
Future<int> computeInIsolate(int input) async {
var receivePort = ReceivePort();
await Isolate.spawn(
_isolateEntry,
[receivePort.sendPort, input],
);
var result = await receivePort.first as int;
receivePort.close();
return result;
}
void _isolateEntry(List<dynamic> args) {
var sendPort = args[0] as SendPort;
var input = args[1] as int;
// 耗时计算在独立Isolate中执行
var result = heavyComputation(input);
sendPort.send(result);
}
int heavyComputation(int n) {
// 模拟CPU密集型计算
var result = 0;
for (var i = 0; i < n; i++) {
result += i;
}
return result;
}
/// Flutter中更简单的方式：compute()
// import 'package:flutter/foundation.dart';
//
// var result = await compute(heavyComputation, 10000000);
// compute自动创建Isolate并传递结果
/// 双向通信Isolate
Future<void> bidirectionalIsolate() async {
var mainReceivePort = ReceivePort();
await Isolate.spawn(
_bidirectionalEntry,
mainReceivePort.sendPort,
);
// 接收Isolate的SendPort
var isolateSendPort = await mainReceivePort.first as SendPort;
// 创建回复端口
var responsePort = ReceivePort();
isolateSendPort.send(['ping', responsePort.sendPort]);
var response = await responsePort.first;
print('主Isolate收到: $response'); // pong
responsePort.close();
mainReceivePort.close();
}
void _bidirectionalEntry(SendPort mainSendPort) {
var receivePort = ReceivePort();
mainSendPort.send(receivePort.sendPort);
receivePort.listen((message) {
if (message is List) {
var data = message[0] as String;
var replyPort = message[1] as SendPort;
if (data == 'ping') {
replyPort.send('pong');
}
}
});
}// ✅ 验证通过：Dart 3.4 + Flutter 3.22
// 文件：lib/async_data_loader.dart
import 'dart:async';
import 'package:flutter/material.dart';
/// 异步状态管理
sealed class AsyncState<T> {
const AsyncState();
}
class AsyncInitial<T> extends AsyncState<T> {
const AsyncInitial();
}
class AsyncLoading<T> extends AsyncState<T> {
final T? previousData;
const AsyncLoading({this.previousData});
}
class AsyncData<T> extends AsyncState<T> {
final T data;
final DateTime loadedAt;
const AsyncData(this.data, {DateTime? loadedAt})
: loadedAt = loadedAt ?? DateTime.now();
}
class AsyncError<T> extends AsyncState<T> {
final Object error;
final StackTrace stackTrace;
final T? previousData;
const AsyncError(this.error, this.stackTrace, {this.previousData});
}
/// 异步数据加载器
class DataLoader<T> {
final Future<T> Function() _fetcher;
final Duration? _refreshInterval;
final Duration _timeout;
StreamController<AsyncState<T>>? _controller;
Timer? _refreshTimer;
AsyncState<T> _state = const AsyncInitial();
DataLoader({
required Future<T> Function() fetcher,
Duration? refreshInterval,
Duration timeout = const Duration(seconds: 10),
}) : _fetcher = fetcher,
_refreshInterval = refreshInterval,
_timeout = timeout;
Stream<AsyncState<T>> get stream {
_controller ??= StreamController<AsyncState<T>>.broadcast(
onCancel: _onCancel,
);
return _controller!.stream;
}
AsyncState<T> get state => _state;
Future<T> load() async {
_emit(AsyncLoading(previousData: _state is AsyncData<T> ? (_state as AsyncData<T>).data : null));
try {
var data = await _fetcher().timeout(_timeout);
_emit(AsyncData(data));
_startAutoRefresh();
return data;
} catch (error, stack) {
_emit(AsyncError(
error,
stack,
previousData: _state is AsyncData<T> ? (_state as AsyncData<T>).data : null,
));
rethrow;
}
}
void _emit(AsyncState<T> state) {
_state = state;
_controller?.add(state);
}
void _startAutoRefresh() {
_refreshTimer?.cancel();
if (_refreshInterval != null) {
_refreshTimer = Timer.periodic(_refreshInterval!, (_) => load());
}
}
void _onCancel() {
_refreshTimer?.cancel();
}
void dispose() {
_refreshTimer?.cancel();
_controller?.close();
}
}
/// Flutter Widget示例
class AsyncDataWidget extends StatefulWidget {
const AsyncDataWidget({super.key});
@override
State<AsyncDataWidget> createState() => _AsyncDataWidgetState();
}
class _AsyncDataWidgetState extends State<AsyncDataWidget> {
late final DataLoader<List<String>> _loader;
StreamSubscription? _subscription;
AsyncState<List<String>> _state = const AsyncInitial();
@override
void initState() {
super.initState();
_loader = DataLoader<List<String>>(
fetcher: _fetchItems,
refreshInterval: const Duration(seconds: 30),
);
_subscription = _loader.stream.listen((state) {
if (mounted) setState(() => _state = state);
});
_loader.load();
}
Future<List<String>> _fetchItems() async {
await Future.delayed(const Duration(seconds: 2));
return List.generate(20, (i) => 'Item ${DateTime.now().second}-$i');
}
@override
void dispose() {
_subscription?.cancel();
_loader.dispose();
super.dispose();
}
@override
Widget build(BuildContext context) {
return Scaffold(
appBar: AppBar(title: const Text('异步数据加载')),
body: switch (_state) {
AsyncInitial() => const Center(child: Text('准备加载')),
AsyncLoading(:final previousData) => Center(
child: previousData != null
? const Column(
mainAxisAlignment: MainAxisAlignment.center,
children: [
CircularProgressIndicator(),
SizedBox(height: 16),
Text('刷新中...'),
],
)
: const CircularProgressIndicator(),
),
AsyncData(:final data) => RefreshIndicator(
onRefresh: () => _loader.load(),
child: ListView.builder(
itemCount: data.length,
itemBuilder: (context, index) => ListTile(
leading: CircleAvatar(child: Text('$index')),
title: Text(data[index]),
),
),
),
AsyncError(:final error, :final previousData) => Center(
child: Column(
mainAxisAlignment: MainAxisAlignment.center,
children: [
const Icon(Icons.error_outline, size: 48, color: Colors.red),
const SizedBox(height: 16),
Text('错误: $error'),
if (previousData != null)
TextButton(
onPressed: () => _loader.load(),
child: const Text('重试'),
),
],
),
),
},
);
}
}不使用async/await，纯用then实现数据加载链：
fetchToken()
.then((token) => fetchProfile(token))
.then((profile) => fetchAvatar(profile.id))
.then((avatar) => displayAvatar(avatar))
.catchError((e) => showError(e));使用Stream实现输入防抖搜索：
class DebouncedSearch {
// 用户停止输入300ms后才触发搜索
// 提示：使用StreamController + asyncMap + Timer
}实现一个最多3个并发任务的下载器：
class ConcurrentDownloader {
final int maxConcurrent;
// 使用Semaphore模式限制并发数
}下一课：集合与泛型——List、Map、Set高级操作，泛型约束，以及Dart 3的类型系统。