Kotlin 流程：异步/等待执行阻塞 UI

Question

我想使用 kotlin 流从 rest 端点异步加载多个数据，并将其显示在回收器视图中。数据定义为

ServiceDto

、

ProductDto

、

CategoryDto

。 ProductDto 具有引用 CategoryDto 的

categoryId

属性。

每个 DTO 都应该异步加载和显示，但 CategoryDto 必须在 ProductDto 之前完成，因为我需要类别名称 + 依赖产品并将其添加到

mutableListOf<Any>

。然后，回收商视图将显示类别名称，后跟相关产品。

产品的存储库实现

fun getProducts(networkId: String) = flow {
    request.getAll("network/${networkId}/products", null)
        .flowOn(Dispatchers.IO)
        .onStart { emit(ApiResult.Loading) }
        .catch { emit(ApiResult.Error(it.message.toString())) }
        .map { ApiResult.Success(it) }
        .also { emitAll(it) }
}

这是加载ServiceDto的实现

private fun loadServices() {
    lifecycleScope.launch {
        async {
            viewModel.getServices(baseViewModel.profile?.networkId ?: "")
        }.await().onEach { result ->
            when (result) {
                is ApiResult.Loading -> {} // Handle loading
                is ApiResult.Error -> {} // Handle error
                is ApiResult.Success -> createServiceChips(result.data) // Create UI elements
            }
        }.collect()
    }
}

这是加载产品和类别的实现

private fun loadCategoriesAndProducts() {
    lifecycleScope.launch {
        val products = async {
            viewModel.getProducts(networkId).shareIn(lifecycleScope, SharingStarted.WhileSubscribed(), 1 // currently no idea what this means ) // Convert to hotflow to start loading data

        }.await()

        async { viewModel.getCategories(networkId) }.await().onEach { categories->
            when (categories) {
                is ApiResult.Loading -> {} // Handle loading
                is ApiResult.Error -> {} // Handle error

                is ApiResult.Success -> {
                    publishCategoryUI(categories.data)
                    collectProducts(products, categories.data)
                }
            }
        }.collect()
    }
}

这就是我收集产品并将类别和产品映射到平面列表的方式。

private suspend fun collectProducts(products: SharedFlow<ApiResult<List<ProductDto>>>, categories: List<CategoryDto>?) = coroutineScope {
    products.onEach{  productResult ->
        when (productResult) {
            is ApiResult.Success -> {
                val productCategoryList = mutableListOf<Any>()
                withContext(Dispatchers.IO) {

                    categories?.forEach { category ->
                        productCategoryList.add(category.name)
                        productResult.data?.filter { product ->
                            product.categoryId == category.id
                        }.let {
                            productCategoryList.addAll(it?.toList() ?: emptyList())
                        }
                    }
                }

                productsAdapter.updateData(productCategoryList) {
                    loadingIndicator.visibility = View.INVISIBLE
                }
            }

            is ApiResult.Error -> // Handle error
            is ApiResult.Loading -> {} // Handle loading
        }
    }.collect()
}

一切正常，但我可以看到，当产品被添加到 recyclerview 时，它会在很短的时间内阻塞 UI。在显示产品之前加载指示器滞后。 （只有 13 个产品）

我怎样才能改进实施或者它是否正确？ Kotlin 协程和流程提供了如此多的可能性，这使得有时很难找到好的/正确的解决方案。

Answer 1

你做的不正确的事情并没有害处，但只会让你的代码变得不必要的复杂：

• 在您的

  getProducts

  函数中，您不需要将流程包装在您从

  flow

  中获得的

  emitAll

  构建器中。放下外

  flow { }

  包装纸和

  .also { emitAll(it) }

  .
• 永远不要做模式

  async { }.await()

  。这与直接调用该 lambda 中的代码没有什么不同。编译器应该在检测到您这样做时向您发出警告。

• onEach { }.collect()

  模式可以缩短为

  .collect { }

  ，或者如我所愿，

  collect()

  可以变成

  launchIn(scope)

  并替换外面的

  scope.launch

  以减少代码嵌套/缩进。
• 在

  collectProducts()

  中，您使用

  coroutineScope

  创建了一个作用域，但从不使用它来运行任何子协程，所以它毫无意义。
• 如果您正在使用 Retrofit、Room 或 Firebase 等知名库中的 Flows，则无需使用

  flowOn

  ，因为它们适当地封装了任何阻塞工作。公共流不应阻塞其下游收集器。同样，您不需要

  withContext

  从这些库中调用挂起函数，因为按照惯例，挂起函数不应该阻塞。

你正在做的不正确和有害的事情：

• 在

  loadCategoriesAndProducts()

  中，您在协程内创建了一个 SharedFlow，因此您有一个热流，在同一个协程内只能有一个收集器。拥有热流的唯一原因是可以多次收集它而无需重新启动它。另一个问题是，只要启动它的 CoroutineScope 还活着，SharedFlow 就会一直存在，如果你创建多个 SharedFlow 就会浪费资源。
• 您正在将此 SharedFlow 传递给另一个收集它的函数。这将创建两个流的乘法，随着项目到达，该流将呈指数级增长。这可以更简单地解决，如下所示。您需要做的只是将这两个流程结合起来。
• 您应该使用

  repeatWithLifecycle

  或

  flowOnLifecycle

  来避免在 Activity 处于屏幕外时进行收集，因为这会浪费资源。
• 在您的

  categories?.forEach { 

  块中，您不必要地进行了嵌套迭代。

如果你想使用 SharedFlow，这可以避免导致必须重新发出新请求，那么你应该在你的 ViewModel 中放置一个函数，将

networkID

提供给 MutableShateFlow，作为另一个 SharedFlow 的基础。您可以在这里私下声明您的两个并行流程，然后将它们公开组合。

// In ViewModel class:
private val _networkId = MutableStateFlow<String?>(null)
var networkId: String?
    get() = _networkId.value
    set(value) { _networkId.value = value }

private val products = _networkId.filterNotNull().distinctUntilChanged()
    .flatMapLatest { networkId ->
        request.getAll("network/${networkId}/products", null)
            .onStart { emit(ApiResult.Loading) }
            .catch { emit(ApiResult.Error(it.message.toString())) }
            .map { ApiResult.Success(it) }
    }

private val categories = _productsNetworkId.filterNotNull().distinctUntilChanged()
    .flatMapLatest { networkId ->
        request.getAll("network/${networkId}/categories", null) // I'm just guessing
            .onStart { emit(ApiResult.Loading) }
            .catch { emit(ApiResult.Error(it.message.toString())) }
            .map { ApiResult.Success(it) }
    }

val categoriesAndProducts: SharedFlow<ApiResult<List<Any>>> =
    products.combine(categories) { p, c ->
        when {
            p is ApiResult.Loading, c is ApiResult.Loading -> ApiResult.Loading
            p is ApiResult.Error -> p
            c is ApiResult.Error -> c
            else -> { // both are Success
                ApiResult.Success( 
                    c.data.orEmpty().flatMap { category ->
                        listOf(category) + p.data.orEmpty().filter { it.categoryId = category.id }
                    }
                )
            }
        }
    }.shareIn(viewModelScope, SharingStarted.WhileSubscribed(5000L), replay = 1)

然后在您的活动中：

private fun loadCategoriesAndProducts() {
    viewModel.networkId = networkId
    viewModel.categoriesAndProducts
        .onEach { 
            when(it) {
                is ApiResult.Loading -> {} // Handle loading
                is ApiResult.Error -> {} // Handle error

                is ApiResult.Success -> {
                    productsAdapter.updateData(productCategoryList) {
                        loadingIndicator.visibility = View.INVISIBLE
                    }
                }
            }
        }
        .flowOn(this, Lifecycle.State.STARTED)
        .launchIn(lifecycleScope)
}

除此之外，使用

MutableList<Any>

是一种代码味道，会造成代码可维护性问题。