android - Kotlin Flow 并行执行两个 API 调用并在每个结果到达时收集它

Question

我正在尝试使用 Kotlin Flows 为我的 API 调用实现缓存然后网络策略。 这是我现在正在尝试的事情

flowOf(
 remoteDataSource.getDataFromCache() // suspending function returning Flow<Data>
   .catch { error -> Timber.e(error) },
 remoteDataSource.getDataFromServer() // suspending function returning Flow<Data>
).flattenConcat().collect {
  Timber.i("Response Received")
}

这里的问题是 collect 仅在 getDataFromServer 返回时调用。我的期望是我应该从缓存中获取第一个事件，然后在几毫秒后从服务器获取第二个事件。在这种情况下，“Response Received”会打印两次，但会一个接一个地打印。

在另一个变体中，“Response Received” 仅在 getDataFromServer() 返回后打印一次。

 remoteDataSource.getDataFromCache() // suspending function returning Flow<Data>
  .catch { error -> Timber.e(error) }
  .flatMapConcat {
    remoteDataSource.getDataFromServer() // suspending function returning Flow<Data>
  }
  .collect {
    Timber.i("Response Received")
  }

我之前使用过 RxJava 的 Flowable.concat() 并且它工作得很好。 Kotlin Flows 中是否有可以模拟这种行为的东西？

Answer 1

Problem here is collect is only called when getDataFromServer returns.

你的设计中的第一个问题是回流功能也是可暂停的。这是两层可悬浮性。函数应该毫无延迟地返回流，并且流本身应该在传入时发出项目。如果您遵循此准则，您的初始代码就已经可以工作了。

按照你编写这些函数的方式，如果你这样写，它们仍然可以工作:

flow<String> {
    emitAll(getCached())
    emitAll(getFromServer())
}

该语句立即完成，返回冷流。当您对其调用 collect 时，它首先调用 getCached() 并发出缓存值，然后调用 getFromServer() 并发出服务器回应。

<小时/>

上述解决方案仅在您使用缓存值后才启动服务器调用。如果您需要同时激活两个流，请使用 flatMapMerge。

假设您解决了上述基本问题并使您的 Flow-returning 函数不暂停，您所需要的就是:

flowOf(getCached(), getFromServer()).flattenMerge()

如果由于某种原因您无法这样做，则必须在每个调用周围添加 emitAll 包装器:

flowOf(
    flow { emitAll(getCached()) }, 
    flow { emitAll(getFromServer()) }
).flattenMerge()