通过领域层连接Bloc：多Bloc如何监听同一数据流？

问题描述 #

我在Repository里实现了一个通过API获取用户信息的Stream方法，多个Bloc会调用这个方法，但现在每个Bloc调用时都会创建独立的Stream，导致API被重复调用（比如2个Bloc就调用2次）。当前的Repository方法和Bloc调用逻辑如下：

Repository代码：

Stream<UserInfo> getUserInfo(
  {required String userKey, required String caller}) async* {
print("Repo Caller is $caller");

  UserInfo response =
      await provider.getUserInfo(
    patientKey: userKey,
  );

  yield response;
}

Bloc调用示例：

// Bloc1
await emit.forEach(
      repo.getUserInfo(userKey: userKey,caller:"bloc1"),
      onData: (UserInfo response) {
        print("bloc1 got data");
        return state.copyWith(...);
      },
    );

// Bloc2
await emit.forEach(
      repo.getUserInfo(userKey: userKey,caller:"bloc2"),
      onData: (UserInfo response) {
        print("bloc2 got data");
        return state.copyWith(...);
      },
    );

输出结果显示API被多次调用：

Repo Caller is bloc1
Repo Caller is bloc2

解决方案 #

问题根源在于每次调用getUserInfo都会创建一个新的async* Stream实例，导致每个Bloc都触发一次API请求。我们可以在Repository中对同一个userKey的Stream进行缓存，并使用广播Stream支持多订阅，从而实现一次API调用，多个Bloc共享结果。

1. 修改Repository实现缓存与广播Stream

class UserRepository {
  final UserProvider provider;
  // 用Map缓存不同userKey对应的Stream实例
  final Map<String, Stream<UserInfo>> _cachedUserStreams = {};

  UserRepository(this.provider);

  Stream<UserInfo> getUserInfo({required String userKey}) {
    // 检查缓存，不存在则创建并缓存
    if (!_cachedUserStreams.containsKey(userKey)) {
      print("Repo: 首次请求用户信息，userKey=$userKey");
      // 将单订阅Stream转为广播Stream，支持多个Bloc同时订阅
      final stream = _fetchUserInfo(userKey).asBroadcastStream();
      _cachedUserStreams[userKey] = stream;
      // 可选：Stream完成后自动移除缓存，下次调用会重新请求
      stream.listen(null, onDone: () {
        _cachedUserStreams.remove(userKey);
      });
    }
    return _cachedUserStreams[userKey]!;
  }

  // 封装实际的API调用逻辑
  Stream<UserInfo> _fetchUserInfo(String userKey) async* {
    final UserInfo response = await provider.getUserInfo(patientKey: userKey);
    yield response;
  }
}

2. Bloc中简化调用

现在Bloc调用时无需传递caller参数，直接调用同一个方法即可：

// Bloc1的监听逻辑
await emit.forEach(
  repo.getUserInfo(userKey: userKey),
  onData: (UserInfo response) {
    print("bloc1 got data");
    return state.copyWith(...);
  },
);

// Bloc2的监听逻辑
await emit.forEach(
  repo.getUserInfo(userKey: userKey),
  onData: (UserInfo response) {
    print("bloc2 got data");
    return state.copyWith(...);
  },
);

3. 预期输出

此时API只会被调用一次，所有Bloc共享同一个结果：

Repo: 首次请求用户信息，userKey=xxx
bloc1 got data
bloc2 got data

可选优化：支持手动刷新

如果需要主动刷新用户信息，可以在Repository中添加刷新方法，清除对应缓存并重新请求：

Future<void> refreshUserInfo(String userKey) async {
  // 移除缓存的Stream
  _cachedUserStreams.remove(userKey);
  // 触发新的请求
  await getUserInfo(userKey).first;
}

内容的提问来源于stack exchange，提问作者Zahra