Future 基础
Future 创建
// 直接创建
Future<String> future = Future.value('Hello');
// 延迟创建
Future<String> delayed = Future.delayed(
Duration(seconds: 2),
() => 'Delayed Hello',
);
// 使用 async
Future<String> asyncFuture() async {
await Future.delayed(Duration(seconds: 1));
return 'Async Hello';
}Future 操作
Future<String> fetchData() async {
// 模拟网络请求
await Future.delayed(Duration(seconds: 2));
return 'Data';
}
// then 链式调用
fetchData()
.then((data) => data.toUpperCase())
.then((upper) => 'Processed: $upper')
.then(print);
// async/await 使用
Future<void> processData() async {
final data = await fetchData();
final upper = data.toUpperCase();
print('Processed: $upper');
}Stream 基础
Stream 创建
// 使用 async*
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<String> stream = controller.stream;
// 广播流
StreamController<String> broadcastController = StreamController.broadcast();Stream 监听
Stream<int> numbers = countStream(5);
// 注意：单订阅流只能监听一次，await for 和 listen 二选一，不能同时用
// 使用 await for
Future<void> printNumbers() async {
await for (final number in numbers) {
print(number);
}
}
// 使用 listen
numbers.listen(
(data) => print('Data: $data'),
onError: (error) => print('Error: $error'),
onDone: () => print('Done'),
);Stream 转换和操作
转换操作
下面每行都用了一个新的 countStream(5)：单订阅流只能被订阅一次，如果 4 个转换共用同一个 numbers，第二次订阅就会抛 Bad state: Stream has already been listened to。
// map 转换
Stream<String> strings = countStream(5).map((n) => 'Number $n');
// where 过滤
Stream<int> evenNumbers = countStream(5).where((n) => n.isEven);
// expand 展开
Stream<int> expanded = countStream(5).expand((n) => [n, n * 2]);
// asyncMap 异步转换
Stream<String> asyncMapped = countStream(5).asyncMap(
(n) async {
await Future.delayed(Duration(milliseconds: 500));
return 'Processed $n';
},
);组合操作
import 'package:async/async.dart'; // StreamGroup 在 async 包里，pubspec 需加 async: ^2.13.0
// 合并流
Stream<int> combined = StreamGroup.merge([
countStream(3),
Stream.periodic(Duration(seconds: 1), (i) => i).take(3),
]);
// 连接流
Stream<int> concatenated = numbers.asyncExpand((n) {
return Stream.fromIterable([n, n * 10]);
});完整示例
import 'dart:async';
import 'package:async/async.dart'; // slices 在 async 包里，pubspec 需加 async: ^2.13.0
class DataService {
final _controller = StreamController<String>();
Stream<String> get dataStream => _controller.stream;
// 模拟数据源
void startGeneratingData() async {
try {
for (int i = 1; i <= 5; i++) {
await Future.delayed(Duration(seconds: 1));
_controller.add('Data packet $i');
}
_controller.close();
} catch (e) {
_controller.addError(e);
}
}
// 模拟数据处理
Stream<String> processData(Stream<String> input) async* {
await for (final data in input) {
yield 'Processed: ${data.toUpperCase()}';
}
}
// 模拟批处理：slices(2) 把流按每 2 个切成一批
Stream<List<String>> batchProcess(Stream<String> input) {
return input.slices(2);
}
Future<void> dispose() async {
await _controller.close();
}
}
class DataProcessor {
final DataService _service;
DataProcessor(this._service);
void start() {
// 开始生成数据
_service.startGeneratingData();
// 处理数据流
_service.dataStream
.transform(StreamTransformer.fromHandlers(
handleData: (data, sink) {
sink.add('Transformed: $data');
},
handleError: (error, stackTrace, sink) {
print('Error: $error');
sink.addError('Processed error: $error');
},
handleDone: (sink) {
print('Stream completed');
sink.close();
},
))
.listen(
print,
onError: (error) => print('Error in stream: $error'),
onDone: () => print('Processing completed'),
);
}
// 批量处理示例
void processBatches() {
_service.dataStream
.slices(2)
.listen(
(batch) => print('Processing batch: $batch'),
onError: (error) => print('Batch error: $error'),
onDone: () => print('Batch processing completed'),
);
}
}
void main() async {
final service = DataService();
final processor = DataProcessor(service);
// 启动处理
processor.start();
// 等待处理完成
await Future.delayed(Duration(seconds: 6));
// 清理资源
await service.dispose();
}高级用法
自定义 StreamTransformer
import 'dart:async';
class ValidationTransformer extends StreamTransformerBase<String, String> {
@override
Stream<String> bind(Stream<String> stream) {
return stream.map((data) {
if (data.isEmpty) {
throw FormatException('Data cannot be empty');
}
return data.trim();
});
}
}
// 使用示例
stream.transform(ValidationTransformer())
.listen(print, onError: print);周期性流
Stream.periodic(Duration(seconds: 1), (i) => i)
.take(5)
.listen(print);最佳实践
- 及时关闭 StreamController
- 合理使用广播流
- 注意内存泄漏
- 处理所有错误
- 使用适当的背压策略
注意事项
- 避免多次监听单订阅流
- 正确处理流的生命周期
- 合理使用转换操作
- 注意异步操作的顺序
- 处理流的取消操作
实战案例：串口数据的广播流
我的 flutter_brightness 项目里，SerialService（lib/services/serial_service.dart）是 Stream 的教科书级用例：
StreamController<String>? _controller;
Stream<String> get dataStream => _controller?.stream ?? const Stream.empty();
bool connect(String portName) {
// ... 打开串口，9600/8N1
_controller = StreamController<String>.broadcast();
_reader = SerialPortReader(_port!);
// 读到数据就 _controller.add(...)
}为什么用 broadcast 而不用普通单订阅 Stream？因为数据的消费者不止一个：UI 层要实时显示光照值，BrightnessMapping 算法要拿数据算目标亮度，DatabaseService 可能还要记日志。单订阅 Stream 只能被 listen 一次——如果 UI 和算法同时订阅，第二个会直接抛异常。所以多消费者的场景，broadcast 是必选项，不是可选项。
还有个细节：get dataStream => _controller?.stream ?? const Stream.empty()。在串口没连接时（_controller 为 null），返回一个空流而不是 null。这样调用方永远拿到一个 Stream<String>，不用写 if (stream != null)——用空对象代替 null 检查，Dart 里 const Stream.empty() 就是干这个的。
Future 和 Stream 的分工在这项目里也很清晰：connect() 返回 bool（Future 语义的一次性结果：连上没连上），dataStream 是 Stream（持续的数据流）。新手常问”什么时候用 Future 什么时候用 Stream”，答案就是：一次性的用 Future，持续的用 Stream。串口连接是一次性的，传感器数据是持续的，就这么简单。
总结
Future 是一次性的结果，Stream 是持续的数据流——串口项目里 connect() 返回 bool、传感器数据走 broadcast 流，就是这个分工。单订阅还是广播、要不要及时 close()，是 Stream 唯二需要操心的事。