高级Dart：学习使用流进行异步编程，了解代码的隔离机制以及事件循环的工作原理

// StreamController.broadcast()用于创建一个可以被任意数量监听器订阅的流。
final controller = StreamController.broadcast();

controller.stream.listen((v) =>> print('监听器1：$v'));
controller.stream.listen((v) =>> print('监听器2：$v');

// 两个监听器都会收到这个值
controller.sink.add('Hello');
// 监听器1：Hello
// 监听器2：Hello

// 使用完流后一定要关闭它，否则会无限占用系统资源。
controller.close();

使用StreamController手动创建数据流

StreamController允许你完全手动控制数据流的生成过程。你可以自行决定何时向数据流中添加数据，何时发送错误信息，以及何时关闭该数据流。通过这种方式，你可以从零开始构建响应式的数据源。

class LocationService {
  // 使用广播功能，让多个组件能够同时接收位置更新信息。
  final _controller = StreamController.broadcast();

  // 只公开数据流本身，而控制器本身保持私有状态，
  // 因此只有这个类才能向其中添加新数据。
  Stream get locationStream => _controller.stream;

  void startTracking() {
    Timer-periodic(const Duration(seconds: 2), (_) {
      final position = Position(lat: 0.3476, lng: 32.5825);
      // 使用sink.add()方法将数据添加到数据流中，
      // 所有正在监听的组件都会立即收到这些数据。
      _controller.sink.add(position);
    });
  }

  void dispose() {
    // 使用完数据流后，一定要关闭它，否则会导致内存泄漏。
    _controller.close();
  }
}

在Flutter中使用StreamBuilder处理数据流

StreamBuilder是Flutter提供的一个组件，它允许你在用户界面中直接使用数据流。每当有新数据到达时，该组件会自动重新构建界面。

StreamBuilder>&(
  stream: firestore
      .collection('messages')
      .snapshots()
      .map((snapshot) => snapshotdocs
          .map((doc) => Message.fromJson(doc.data()))
          .ToList()),
  builder: (context, snapshot) {
    if (snapshot.connectionState == ConnectionState.waiting) {
      return const CircularProgressIndicator();
    }

    if (snapshot.hasError) {
      return Text('错误：${snapshot.error}`);
    }

    if (!snapshot.hasData || snapshot.data!.isEmpty) {
      return const Text('目前还没有消息');
    }

    return ListView.builder(
      itemCount: snapshot.data!.length,
      itemBuilder: (context, index) {
        return MessageBubble(message: snapshot.data![index]);
      },
    );
  },
)

在dispose方法中务必取消数据流订阅

这是Flutter应用程序中最常见的内存泄漏问题之一，其根本原因在于人们对数据流的原理理解不够透彻。

当某个组件订阅了数据流后，相应的回调函数会一直处于活跃状态。即使该组件已经被从界面中移除，如果订阅仍然存在，这些回调函数仍会继续执行，导致setState方法在dispose之后被调用，从而使得那些本应被释放的对象继续占用内存。

class _ChatScreenState extends State {
  StreamSubscription? _subscription;

  @override
  void initState() {
    super.initState();
    _subscription = messageStream.listen((message) {
      if (mounted) setState(() => messages.add(message));
    });
  }

  @override
  void dispose() {
    // 使用cancel()方法取消数据流订阅，
    // 这样就可以防止回调函数在组件被移除后继续执行。
    _subscription?.cancel();
    super.dispose();
  }
}

流转换器与高级流控制

一旦你了解了流的概念，就会很快发现，原始的流数据很少能直接满足你的需求。你需要对数据进行处理、进行转换、抑制频繁的数据发送，或者将多个流合并起来。这时，流操作符和`StreamTransformer`就派上用场了。

Dart的`Stream`类提供了丰富的内置转换方法：

final stream = countStream();

// map — 在数据传达到监听器之前对其进行转换
stream
    .map((number) => number * 2)
    .listen(print); // 输出：2, 4, 6, 8, 10

// where — 过滤掉不符合条件的数据
stream
    .where((number) => number.isEven)
    .listen(print); // 输出：2, 4

// take — 只输出前N个数据，然后停止发送
stream
    .take(3)
    .listen(print); // 输出：1, 2, 3

// skip — 忽略前N个数据
stream
    .skip(2)
    .listen(print); // 输出：3, 4, 5

// distinct — 只有当数据与上一次输入的数据不同时才输出
Stream.fromIterable([1, 1, 2, 2, 3])
    .distinct()
    .listen(print); // 输出：1, 2, 3

对于更复杂的转换需求，你可以自定义`StreamTransformer`。当内置操作符无法满足你的需求时，就可以使用这种方法——比如当你需要以某种需要在数据发送过程中维护状态的方式来处理数据时。

// 这个StreamTransformer只会输出超过特定阈值的值，并在每个值前面添加标签
StreamTransformer aboveThreshold(int threshold) {
  return StreamTransformer.fromHandlers(
    handleData: (value, sink) {
      // sink.add()会将转换后的值发送下去。
      // 如果不调用sink.add()，这个值就会被忽略。
      if (value > threshold) {
        sink.add('超过阈值：$value');
      }
    },
    handleError: (error, stackTrace, sink) {
      // 将错误原封不动地发送下去。
      sink.addError(error, stackTrace);
    },
    handleDone: (sink) {
      // 当输入流关闭时，关闭输出流。
      sink.close();
    },
  );
}

// 使用方法
countStream()
    .transform(aboveThreshold(3))
    .listen(print); // 输出：超过阈值：4, 超过阈值：5

在Flutter中利用流实现去抖动功能

在Flutter应用中，最实用的流处理模式之一就是为搜索框实现去抖动功能。如果不使用去抖动机制，每次用户按键都会触发一次API调用；而通过去抖动，可以等到用户停止输入后再进行请求。

class _SearchScreenState extends State {
  final _searchController = TextEditingController();
  final _searchStream = StreamController();
  StreamSubscription? _subscription;
  List _results = [];

  @override
  void initState() {
    super.initState();

    _subscription = _searchStream.stream
        // 在用户最后一次按键后等待300毫秒再发送数据。
        // 如果在300毫秒内又有新的输入，计时器会重新开始计算。
        // 这样就可以避免每次按键都触发API调用。
        .asyncExpand((query) async* {
          await Future.delayed(const Duration(milliseconds: 300));
          yield query;
        })
        // 忽略重复的查询请求——如果用户又输入了相同的内容，就没有必要再次发送请求。
        .distinct()
        // 对每个查询请求，都会调用API并获取结果。
        // 如果在之前的请求还未完成之前又有新的请求到来，asyncMap会取消之前的请求。
        .asyncMap((query) => _repository.search(query))
        .listen((results) {
          if (mounted) setState(() => _results = results);
        });

    _searchController.addListener(() {
      _searchStream.add(_searchController.text);
    });
  }

  @override
  void dispose() {
    _searchController.dispose();
    _subscription?.cancel();
    _searchStream.close();
    superdispose();
  }
}

隔离执行：摆脱单线程的限制

Dart是单线程语言，但这并不意味着你永远只能使用一个线程。Dart的“隔离执行”机制允许你在完全独立的线程中运行代码——不过这与其他语言中的线程机制存在一个重要区别。

在大多数编程语言中，线程会共享内存。两个线程可以同时读写同一个变量，这就容易导致竞争条件，因此需要通过锁定机制来避免这些问题。

然而Dart的隔离执行机制根本不会共享内存。每个隔离执行单元都有自己独立的内存堆。两个隔离执行单元之间只能通过传递消息来进行通信——这就好比是通过墙上的缝隙传递信息，而不是共用一块白板。

这种设计使得隔离执行单元具有很高的安全性。由于不存在共享内存，因此也就不会发生竞争条件；每个隔离执行单元都能完全掌控自己的数据。

主隔离执行单元                    工作线程隔离执行单元
─────────────────               ─────────────────
拥有独立的内存堆                 拥有独立的内存堆
拥有独立的事件循环                  拥有独立的事件循环
负责UI渲染                    负责繁重计算任务
接收用户输入                      无法访问UI界面
│                               │
│──── 发送数据 ──────────────→│
│                               │ （各自独立运行）
│←─── 接收结果 ──────────│

何时真正需要使用隔离执行单元

关键的区别在于任务是依赖于CPU的计算还是依赖于I/O操作的：

• 依赖I/O的操作：比如等待网络响应、读取文件等——在这种情况下可以直接使用await。在等待期间，CPU处于空闲状态，因此事件循环依然可以正常运行。

• 依赖CPU的计算任务：比如进行复杂的数据处理、解析大型文件等——这类任务需要使用隔离执行单元。因为在这些操作过程中CPU会一直保持忙碌状态，所以await无法起到帮助作用。

如果解析API响应需要200毫秒的时间，那么使用await也无法节省时间——在这200毫秒里，事件循环仍然会被阻塞。因此，这类任务必须被放到单独的隔离执行单元中来处理。

Isolate.run()”——现代化的解决方案

// 在你的代码库中：
Future> getUsers() async {
  // 第一步——网络请求属于I/O操作，可以使用await等待。
  final response = await dio.get('/users');

  // 第二步——解析大量用户数据需要依赖CPU计算，因此需要使用隔离执行单元。
  final users = await Isolate.run(() {
    final data = response.data as List;
    return data
        .map((json) => User.fromJson(json as Map

传递的函数必须是顶层函数或静态函数，
不能是闭包，因为能够捕获状态的闭包无法在隔离执行环境之间传递。
List parseUsers(dynamic data) {
  return (data as List)
      .map((json) => User.fromJson(json as Map

void main() async {
  // ReceivePort用于接收来自工作线程的消息。
  final receivePort = ReceivePort();

  // 创建工作线程隔离执行环境，并为其提供SendPort，以便它能够向我们发送消息。
  await Isolate.spawn(
    workerFunction,
    receivePort.sendPort,
  );

  // 监听工作线程发送的消息。
  receivePort.listen((message) {
    print('主线程收到消息：$message');
  });
}

// 这个函数完全在工作线程隔离执行环境中运行，
// 它拥有自己独立的内存堆，与主线程完全隔离开来，
// 因此无法直接访问主线程中的任何变量。
void workerFunction(SendPort sendPort) {
  for (int i = 0; i < 5; i++) {
    // sendPort.send()用于向主线程发送消息。
    // 这些消息是会被复制传递的，而不是共享内存，
    // 因此不存在共享内存的问题。
    sendPort.send('已处理的项目编号 $i');
  }
}