开篇：异步是 Flutter 的 DNA
Dart 语言是单线程事件循环模型，与 JavaScript 类似。这种设计避免了多线程的竞态条件问题，但也意味着所有代码默认在主 isolate 上执行——如果某个操作阻塞了事件循环，整个 UI 都会卡住。因此，在 Flutter 中编写高效的异步代码不仅是性能优化，更是用户体验的基础。
从 Future 的链式调用到 async/await 的语法糖，从 Stream 的响应式数据流到 Isolate 的真正并行计算，Dart 提供了丰富的异步工具。而在网络通信层面，从简单的 http 调用到 Dio 的强大封装，从 RESTful API 到 GraphQL 查询，从 WebSocket 实时通信到离线优先的本地缓存策略——Flutter 的网络生态足够成熟，可以支撑从小型应用到大型平台的各种需求。
一、Dart 异步模型深度解析
1.1 事件循环（Event Loop）
// Dart 单线程事件循环的核心流程：
// 1. 执行 main() 中的同步代码
// 2. 检查 Microtask 队列，全部执行
// 3. 检查 Event 队列，执行一个事件
// 4. 重复 2-3
void main() {
print('1 - 同步代码');
Future.microtask(() => print('2 - Microtask'));
Future(() => print('3 - Event (Future)'));
Future.delayed(Duration.zero, () => print('4 - Delayed Future'));
print('5 - 同步代码结束');
}
// 输出顺序：1 → 5 → 2 → 3 → 4
优先级：同步代码 > Microtask > Event。理解这个顺序对预测代码执行时序至关重要。
一句话总结：Dart 事件循环的同步→Microtask→Event 优先级规则，是理解和调试异步行为的基础。
1.2 async/await 语法糖
// 原始 Future 链式调用
Future<String> fetchData() {
return http.get(Uri.parse('/api/data'))
.then((response) {
if (response.statusCode == 200) return response.body;
throw Exception('请求失败');
})
.then((body) => jsonDecode(body))
.then((data) => data['title'] as String)
.catchError((error) => '默认值');
}
// async/await 改写（更易读）
Future<String> fetchData() async {
try {
final response = await http.get(Uri.parse('/api/data'));
if (response.statusCode != 200) throw Exception('请求失败');
final data = jsonDecode(response.body);
return data['title'] as String;
} catch (e) {
return '默认值';
}
}
// 并行等待多个 Future
Future<Map<String, dynamic>> fetchDashboard() async {
final results = await Future.wait([
fetchUserProfile(),
fetchNotifications(),
fetchRecentOrders(),
]);
return {
'profile': results[0],
'notifications': results[1],
'orders': results[2],
};
}
一句话总结：async/await 不是新机制，而是 Future 的语法糖，它让异步代码的阅读和编写像同步代码一样自然。
1.3 Stream 响应式编程
import 'dart:async';
// Stream 控制器
class ChatService {
final _messageController = StreamController<Message>.broadcast();
Stream<Message> get messages => _messageController.stream;
void sendMessage(Message msg) {
_messageController.add(msg);
}
void dispose() {
_messageController.close();
}
}
// Stream 变换
void demoStream() async {
final stream = Stream.fromIterable([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]);
await stream
.map((n) => n * 2)
.where((n) => n > 5)
.take(3)
.forEach(print); // 12, 14, 16
// 监听 WebSocket 消息
final wsStream = webSocketChannel.stream;
wsStream
.map((data) => jsonDecode(data))
.where((json) => json['type'] == 'message')
.cast<Map<String, dynamic>>()
.map((json) => Message.fromJson(json))
.listen(
(message) => displayMessage(message),
onError: (e) => logError(e),
onDone: () => showDisconnected(),
);
}
// 自定义 Stream（async* / yield）
Stream<int> countdown(int from) async* {
for (var i = from; i >= 0; i--) {
await Future.delayed(Duration(seconds: 1));
yield i; // 产生一个值
}
yield -1; // 完成信号
}
// yield* 委托给另一个 Stream
Stream<int> doubleCountdown(int from) async* {
yield* countdown(from).map((n) => n * 2);
}
一句话总结：Stream 是 Dart 对"异步数据序列"的抽象，通过 map/where/take 等操作符可以像处理集合一样处理异步事件流。
二、Isolate：真正的并行计算
2.1 为什么要用 Isolate？
// ❌ 在主 isolate 执行耗时计算 → UI 卡顿
void heavyComputation() {
var sum = 0;
for (var i = 0; i < 100000000; i++) {
sum += i;
}
return sum;
}
// ✅ 使用 compute() 在后台 isolate 执行
Future<int> heavyComputationAsync() async {
return compute((message) {
var sum = 0;
for (var i = 0; i < 100000000; i++) {
sum += i;
}
return sum;
}, null);
}
// compute() 自动处理 isolate 创建、通信、销毁
// 适用场景：图片处理、JSON 解析、大文件处理、复杂计算
2.2 手动管理 Isolate
import 'dart:isolate';
class ImageProcessor {
late SendPort _sendPort;
late Isolate _isolate;
Future<void> init() async {
final receivePort = ReceivePort();
_isolate = await Isolate.spawn(
_imageProcessorEntry,
receivePort.sendPort,
);
_sendPort = await receivePort.first;
}
Future<Uint8List> processImage(Uint8List imageData) async {
final receivePort = ReceivePort();
_sendPort.send({
'replyPort': receivePort.sendPort,
'image': imageData,
});
return await receivePort.first;
}
void dispose() {
_isolate.kill();
}
static void _imageProcessorEntry(SendPort mainSendPort) {
final receivePort = ReceivePort();
mainSendPort.send(receivePort.sendPort);
receivePort.listen((message) async {
final replyPort = message['replyPort'] as SendPort;
final image = message['image'] as Uint8List;
// 耗时图像处理
final processed = await _applyFilters(image);
replyPort.send(processed);
});
}
static Future<Uint8List> _applyFilters(Uint8List image) async {
// 图像处理逻辑
return image;
}
}
一句话总结：Isolate 让 Dart 突破了单线程限制，但通信开销较大——compute() 适合一次性任务，手动管理适合需要长期保持的后台工作线程。
三、Dio 深度实战
Dio 是 Flutter 生态中最强大的 HTTP 客户端，提供了拦截器、全局配置、请求取消、进度监听等高级功能。
3.1 基础封装
import 'package:dio/dio.dart';
class ApiClient {
late final Dio _dio;
ApiClient({String baseUrl = 'https://api.example.com'}) {
_dio = Dio(BaseOptions(
baseUrl: baseUrl,
connectTimeout: Duration(seconds: 10),
receiveTimeout: Duration(seconds: 30),
headers: {'Content-Type': 'application/json'},
));
_setupInterceptors();
}
void _setupInterceptors() {
// 请求拦截器：添加 Token
_dio.interceptors.add(InterceptorsWrapper(
onRequest: (options, handler) async {
final token = await _getToken();
if (token != null) {
options.headers['Authorization'] = 'Bearer $token';
}
print('→ REQUEST: ${options.method} ${options.path}');
handler.next(options);
},
onResponse: (response, handler) {
print('← RESPONSE [${response.statusCode}]: ${response.requestOptions.path}');
handler.next(response);
},
onError: (error, handler) async {
if (error.response?.statusCode == 401) {
// Token 过期，尝试刷新
final refreshed = await _refreshToken();
if (refreshed) {
// 重试原请求
final retry = await _dio.fetch(error.requestOptions);
handler.resolve(retry);
return;
}
}
handler.next(error);
},
));
// 日志拦截器（仅开发环境）
if (kDebugMode) {
_dio.interceptors.add(LogInterceptor(
requestBody: true,
responseBody: true,
));
}
}
Future<T> get<T>(String path, {
Map<String, dynamic>? queryParameters,
Options? options,
}) async {
final response = await _dio.get<T>(path,
queryParameters: queryParameters,
options: options,
);
return response.data as T;
}
Future<T> post<T>(String path, {dynamic data}) async {
final response = await _dio.post<T>(path, data: data);
return response.data as T;
}
// 文件下载
Future<void> downloadFile(
String url,
String savePath, {
void Function(int received, int total)? onProgress,
CancelToken? cancelToken,
}) async {
await _dio.download(
url,
savePath,
onReceiveProgress: onProgress,
cancelToken: cancelToken,
);
}
}
3.2 完整 Repository 层
// 统一响应封装
class ApiResponse<T> {
final bool success;
final T? data;
final String? message;
final int? code;
ApiResponse({required this.success, this.data, this.message, this.code});
factory ApiResponse.fromJson(
Map<String, dynamic> json,
T Function(dynamic) converter,
) {
return ApiResponse(
success: json['success'] ?? false,
data: json['data'] != null ? converter(json['data']) : null,
message: json['message'],
code: json['code'],
);
}
}
// Repository 实现
class UserRepository {
final ApiClient _client;
UserRepository(this._client);
Future<User> getUser(String id) async {
final response = await _client.get<Map<String, dynamic>>('/users/$id');
final apiResponse = ApiResponse.fromJson(response!, (d) => User.fromJson(d));
if (!apiResponse.success) throw ApiException(apiResponse.message);
return apiResponse.data!;
}
Future<List<User>> searchUsers(String query) async {
final response = await _client.get<Map<String, dynamic>>(
'/users',
queryParameters: {'q': query},
);
final apiResponse = ApiResponse.fromJson(
response!,
(d) => (d as List).map((u) => User.fromJson(u)).toList(),
);
return apiResponse.data ?? [];
}
Future<User> updateProfile(String userId, UpdateProfileRequest request) async {
final response = await _client.post<Map<String, dynamic>>(
'/users/$userId',
data: request.toJson(),
);
return User.fromJson(ApiResponse.fromJson(response!, (d) => d).data!);
}
}
3.3 请求取消
class SearchController extends ChangeNotifier {
final _api = ApiClient();
CancelToken? _cancelToken;
List<SearchResult> _results = [];
List<SearchResult> get results => _results;
Future<void> search(String query) async {
// 取消上一个请求
_cancelToken?.cancel('新搜索请求');
_cancelToken = CancelToken();
try {
final response = await _api.get('/search',
queryParameters: {'q': query},
cancelToken: _cancelToken,
);
_results = (response as List).map((r) => SearchResult.fromJson(r)).toList();
notifyListeners();
} on DioException catch (e) {
if (CancelToken.isCancel(e)) return; // 正常取消，忽略
rethrow;
}
}
@override
void dispose() {
_cancelToken?.cancel('Controller disposed');
super.dispose();
}
}
一句话总结：Dio 的拦截器架构将认证、日志、重试等横切关注点从业务代码中剥离，配合 cancelToken 实现防抖搜索，是 Flutter 网络层的工业标准。
四、REST API 最佳实践
4.1 数据模型层
import 'package:json_annotation/json_annotation.dart';
part 'models.g.dart'; // 由 build_runner 生成
@JsonSerializable()
class User {
final String id;
final String name;
final String? email;
@JsonKey(name: 'created_at')
final DateTime createdAt;
@JsonKey(defaultValue: [])
final List<String> roles;
const User({
required this.id,
required this.name,
this.email,
required this.createdAt,
required this.roles,
});
factory User.fromJson(Map<String, dynamic> json) => _$UserFromJson(json);
Map<String, dynamic> toJson() => _$UserToJson(this);
User copyWith({String? name, String? email, List<String>? roles}) {
return User(
id: id,
name: name ?? this.name,
email: email ?? this.email,
createdAt: createdAt,
roles: roles ?? this.roles,
);
}
}
@JsonSerializable()
class ApiError {
final String code;
final String message;
final Map<String, dynamic>? details;
ApiError({required this.code, required this.message, this.details});
factory ApiError.fromJson(Map<String, dynamic> json) => _$ApiErrorFromJson(json);
@override
String toString() => 'ApiError($code): $message';
}
4.2 统一错误处理
class ApiException implements Exception {
final String message;
final int? statusCode;
final ApiError? error;
ApiException(this.message, {this.statusCode, this.error});
factory ApiException.fromDio(DioException e) {
switch (e.type) {
case DioExceptionType.connectionTimeout:
case DioExceptionType.sendTimeout:
case DioExceptionType.receiveTimeout:
return ApiException('请求超时，请检查网络', statusCode: 408);
case DioExceptionType.connectionError:
return ApiException('网络连接失败', statusCode: 0);
case DioExceptionType.badResponse:
final status = e.response?.statusCode;
final data = e.response?.data;
if (data != null && data is Map<String, dynamic>) {
return ApiException(
data['message'] ?? '请求失败',
statusCode: status,
error: ApiError.fromJson(data),
);
}
return ApiException('请求失败 (HTTP $status)', statusCode: status);
default:
return ApiException('未知错误: ${e.message}', statusCode: 0);
}
}
bool get isNetworkError => statusCode == 0 || statusCode == 408;
bool get isAuthError => statusCode == 401 || statusCode == 403;
}
// 全局错误处理 Widget
class AsyncValueWidget<T> extends StatelessWidget {
final AsyncValue<T> value;
final Widget Function(T data) data;
final Widget? loading;
final Widget Function(Object error, StackTrace? stack)? error;
const AsyncValueWidget({
super.key,
required this.value,
required this.data,
this.loading,
this.error,
});
@override
Widget build(BuildContext context) {
return value.when(
data: data,
loading: () => loading ?? const Center(child: CircularProgressIndicator()),
error: (e, st) => error?.call(e, st) ?? _buildError(context, e),
);
}
Widget _buildError(BuildContext context, Object error) {
final message = error is ApiException
? error.message
: '发生未知错误';
return Center(child: Text(message, style: TextStyle(color: Colors.red)));
}
}
一句话总结：清晰的数据模型层 + 统一的错误处理机制，是构建可维护网络层的基石。
五、WebSocket 实时通信
import 'package:web_socket_channel/web_socket_channel.dart';
import 'package:web_socket_channel/io.dart';
class ChatWebSocketService {
WebSocketChannel? _channel;
final _messageController = StreamController<ChatMessage>.broadcast();
final _statusController = StreamController<ConnectionStatus>.broadcast();
Stream<ChatMessage> get messages => _messageController.stream;
Stream<ConnectionStatus> get status => _statusController.stream;
void connect(String roomId, String token) {
_disconnect();
_statusController.add(ConnectionStatus.connecting);
try {
_channel = IOWebSocketChannel.connect(
'wss://chat.example.com/rooms/$roomId',
headers: {'Authorization': 'Bearer $token'},
);
_channel!.stream.listen(
(data) {
_statusController.add(ConnectionStatus.connected);
final json = jsonDecode(data);
_messageController.add(ChatMessage.fromJson(json));
},
onError: (error) {
_statusController.add(ConnectionStatus.error);
_scheduleReconnect(roomId, token);
},
onDone: () {
_statusController.add(ConnectionStatus.disconnected);
_scheduleReconnect(roomId, token);
},
);
// 启动心跳
_startHeartbeat();
} catch (e) {
_statusController.add(ConnectionStatus.error);
_scheduleReconnect(roomId, token);
}
}
void sendMessage(String content) {
_channel?.sink.add(jsonEncode({
'type': 'message',
'content': content,
'timestamp': DateTime.now().toIso8601String(),
}));
}
void _startHeartbeat() {
Timer.periodic(Duration(seconds: 30), (_) {
_channel?.sink.add(jsonEncode({'type': 'ping'}));
});
}
void _scheduleReconnect(String roomId, String token) {
Future.delayed(Duration(seconds: 5), () => connect(roomId, token));
}
void _disconnect() {
_channel?.sink.close();
_channel = null;
}
void dispose() {
_disconnect();
_messageController.close();
_statusController.close();
}
}
enum ConnectionStatus { connecting, connected, disconnected, error }
一句话总结：WebSocket 连接管理需要处理连接、心跳、重连、断开等全生命周期，封装为服务类是最佳实践。
六、离线优先架构
import 'package:connectivity_plus/connectivity_plus.dart';
import 'package:hive/hive.dart';
class OfflineFirstRepository<T> {
final ApiClient _api;
final Box<T> _localBox;
final String apiEndpoint;
OfflineFirstRepository(this._api, this._localBox, this.apiEndpoint);
Future<List<T>> getAll() async {
final connectivity = await Connectivity().checkConnectivity();
if (connectivity == ConnectivityResult.none) {
// 离线：返回本地缓存
return _localBox.values.toList();
}
try {
// 在线：获取远程数据并缓存
final remoteData = await _api.get<List<dynamic>>(apiEndpoint);
final items = remoteData!.map((d) => _deserialize(d)).toList();
// 更新本地缓存
await _localBox.clear();
for (var i = 0; i < items.length; i++) {
await _localBox.put(i, items[i]);
}
return items;
} catch (e) {
// 请求失败次选本地数据
return _localBox.values.toList();
}
}
Future<T> create(T item) async {
// 先生成本地 ID
final localId = DateTime.now().millisecondsSinceEpoch.toString();
await _localBox.put(localId, item);
// 尝试同步到服务器
_syncToServer();
return item;
}
Future<void> _syncToServer() async {
final pending = _localBox.values.toList();
final connectivity = await Connectivity().checkConnectivity();
if (connectivity == ConnectivityResult.none) return;
for (final item in pending) {
try {
await _api.post(apiEndpoint, data: _serialize(item));
// 标记同步成功
} catch (e) {
// 保留在本地，下次重试
break;
}
}
}
T _deserialize(dynamic data) => throw UnimplementedError();
Map<String, dynamic> _serialize(T item) => throw UnimplementedError();
}
一句话总结：离线优先不是不依赖网络，而是在网络不可用时 gracefully 降级到本地缓存，确保核心功能始终可用。
七、文件上传与下载
// 多文件上传
Future<void> uploadFiles(List<XFile> files) async {
final formData = FormData();
for (final file in files) {
formData.files.add(MapEntry(
'attachments',
await MultipartFile.fromFile(file.path, filename: file.name),
));
}
final response = await dio.post('/upload',
data: formData,
onSendProgress: (sent, total) {
final progress = sent / total;
print('上传进度: ${(progress * 100).toStringAsFixed(1)}%');
},
);
}
// 断点续传下载
Future<void> downloadWithResume(String url, String savePath) async {
final file = File(savePath);
var startByte = 0;
if (await file.exists()) {
startByte = await file.length();
}
final cancelToken = CancelToken();
await dio.download(
url,
savePath,
cancelToken: cancelToken,
options: Options(headers: {'Range': 'bytes=$startByte-'}),
onReceiveProgress: (received, total) {
if (total != -1) {
print('下载进度: ${((received + startByte) / (total + startByte) * 100).toStringAsFixed(1)}%');
}
},
);
}
FAQ
Q1: Future.wait 中一个失败会怎样？
默认情况下，任何一个 Future 失败都会导致整个 wait 失败。使用 Future.wait(futures, eagerError: true) 可尽早失败，或使用 Result 类型包装每个 Future。
Q2: Stream 的 listen 可以多次调用吗？
取决于 Stream 类型：
- 单订阅 Stream（默认）只能 listen 一次
- 广播 Stream（
.asBroadcastStream()或StreamController.broadcast()）可多次 listen
Q3: 如何避免内存泄漏？
始终取消 Stream 订阅和关闭控制器：
final subscription = stream.listen(...);
// 在 dispose 中：
subscription.cancel();
controller.close();
Q4: Dio 的拦截器执行顺序？
请求：按添加顺序执行（FIFO）
响应：按添加顺序的逆序执行（LIFO）
错误：按添加顺序的逆序执行
Q5: compute() 的参数有什么限制？
传递给 compute 的函数和参数必须是顶级函数或静态方法，且参数必须是可序列化的——不能包含 BuildContext、StreamController 等不可传递的对象。
Q6: WebSocket 和 SSE 怎么选？
- 双向实时通信（聊天、游戏、协同编辑）→ WebSocket
- 服务器单向推送（股票行情、通知、日志流）→ SSE（更轻量，自动重连，兼容 HTTP 缓存）
Q7: 如何处理网络请求缓存？
Dio 内置缓存拦截器：dio.interceptors.add(DioCacheInterceptor(options: cacheOptions))，配合 CachePolicy.request/CachePolicy.refresh 灵活控制缓存策略。
相关阅读
- https://plumephp.com/flutter-local-storage/ — 本地存储与持久化
- https://plumephp.com/graphql-fundamentals/ — GraphQL 基础
- https://plumephp.com/dart-basics/ — Dart 异步编程语言特性
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。