Coroutine Flow结合Retrofit2突发取消问题排查与解决

触发SwipeRefreshLayout时偶尔出现<-- HTTP FAILED: java.io.IOException: Canceled错误，未手动编写取消请求的代码，完全依赖生命周期组件处理，怀疑是协程被取消导致，求问题原因及解决办法。

相关代码片段 #

1. Fragment的onViewCreated代码 #

with(viewModel) {
    viewLifecycleOwner.collectShared(newsState, ::onNewsStateChanged)
    // Initial fetch
    fetchNews(
        "some_data",
        false
    )

    swipeRefreshLayout.setOnRefreshListener {
        // Reload data
        viewModel.fetchNews("some_data", true)
    }
}

2. 扩展函数代码 #

inline fun <T : Any, L : SharedFlow<T>> LifecycleOwner.collectShared(
    sharedFlow: L,
    crossinline function: (T) -> Unit,
    lifecycleState: Lifecycle.State = Lifecycle.State.STARTED
) {
    lifecycleScope.launch {
        repeatOnLifecycle(lifecycleState) {
            sharedFlow.collect { t -> function(t) }
        }
    }
}

fun <T : Any> defaultMutableSharedFlow() =
    MutableSharedFlow<T>(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)

3. ViewModel代码 #

private val _newsState = defaultMutableSharedFlow<NewsState>()

val newsState = _newsState.asSharedFlow()

fun fetchNews(category: String, isReload: Boolean) {

    viewModelScope.launch {

        getNewsUseCase(
            isReload,
            AppConfig.remote.newsLink,
            AppConfig.remote.newsField,
            category,
            if (isReload)
                getNewsUseCase.getPageNumber().minus(1)
            else
                getNewsUseCase.getPageNumber(),
        ).onEach {

            when (it) {

                is RequestStatus.Loading -> {
                    _newsState.tryEmit(NewsState.FetchLoading(it.data))
                }

                is RequestStatus.Success -> {
                    _newsState.tryEmit(NewsState.FetchSuccess(it.data))
                }

                is RequestStatus.Canceled -> {
                    // No operation is needed
                }

                is RequestStatus.Failed -> {
                    _newsState.tryEmit(NewsState.FetchFailed(it.message, it.data))
                }

            }

        }.collect()

        _lastSelectedCategory = category

    }

}

4. Repository代码 #

override fun fetchNews(
    isReload: Boolean,
    baseUrl: String,
    query: String,
    category: String,
    page: Int
) = flow {

    emit(RequestStatus.Loading())

    var domainModel = mapper.mapToDomainModelList(dao.getNewsByCategoryId(category, page))

    try {

        val cacheList = getProcessedNewsList(domainModel)

        // Fetching data from remote can take longer
        // Showing the cache first if available
        emit(RequestStatus.Loading(cacheList))

        val dtoModel = service.getNewsItems(
            baseUrl + "posts",
            query,
            category,
            page.toString()
        )

        val entities = mapper.mapToEntityModelList(dtoModel)
        entities.forEach {
            it.category = category
        }

        // Updating cache
        dao.insert(entities)

        // Get the updated cache
        domainModel = mapper.mapToDomainModelList(dao.getNewsByCategoryId(category, page))

        val freshData = getProcessedNewsList(domainModel)

        // New page was fetched, if API response returns empty list then retain the page number
        if (domainModel.isNotEmpty() && isReload.not())
            _pageNumber = _pageNumber.plus(1)

        emit(RequestStatus.Success(freshData))

    } catch (e: HttpException) {

        val cacheDomainList = getProcessedNewsList(domainModel)

        // Support pagination for offline
        if (domainModel.isNotEmpty() && isReload.not())
            _pageNumber = _pageNumber.plus(1)

        emit(RequestStatus.Failed(e, cacheDomainList))

    } catch (e: IOException) {

        val cacheDomainList = getProcessedNewsList(domainModel)

        // Support pagination for offline
        if (domainModel.isNotEmpty() && isReload.not())
            _pageNumber = _pageNumber.plus(1)

        emit(RequestStatus.Failed(e, cacheDomainList))

    }

}

1. 重复请求导致旧协程被取消：每次触发下拉刷新都会调用viewModel.fetchNews并启动新协程，若前一次请求仍在进行中，网络库（如Retrofit）内部会自动取消旧请求以避免并发冲突，最终抛出IOException: Canceled错误。
2. 未区分主动取消与真实IO错误：Repository中未捕获CancellationException，导致协程主动取消的信号被转换成了IO异常，干扰了错误判断。

1. 主动管理请求协程生命周期 #

在ViewModel中维护当前请求的协程Job，每次发起新请求前取消旧请求，避免并发冲突：

private var fetchJob: Job? = null

fun fetchNews(category: String, isReload: Boolean) {
    // 取消未完成的旧请求
    fetchJob?.cancel()
    fetchJob = viewModelScope.launch {
        // 原有fetchNews逻辑
        getNewsUseCase(
            isReload,
            AppConfig.remote.newsLink,
            AppConfig.remote.newsField,
            category,
            if (isReload) getNewsUseCase.getPageNumber().minus(1) else getNewsUseCase.getPageNumber()
        ).onEach {
            when (it) {
                is RequestStatus.Loading -> _newsState.tryEmit(NewsState.FetchLoading(it.data))
                is RequestStatus.Success -> _newsState.tryEmit(NewsState.FetchSuccess(it.data))
                is RequestStatus.Canceled -> {}
                is RequestStatus.Failed -> _newsState.tryEmit(NewsState.FetchFailed(it.message, it.data))
            }
        }.collect()
        _lastSelectedCategory = category
    }
}

2. 区分协程取消与真实IO错误 #

在Repository的流中单独捕获CancellationException，避免将主动取消误判为IO错误：

override fun fetchNews(...) = flow {
    try {
        // 原有流逻辑
    } catch (e: CancellationException) {
        // 协程主动取消，无需处理，直接抛出让协程正常结束
        throw e
    } catch (e: HttpException) {
        // 原有Http异常处理逻辑
    } catch (e: IOException) {
        // 原有IO异常处理逻辑
    }
}

3. 控制下拉刷新触发状态 #

通过NewsState控制下拉刷新控件的状态，避免重复触发请求：

private fun onNewsStateChanged(state: NewsState) {
    when(state) {
        is NewsState.FetchLoading -> swipeRefreshLayout.isRefreshing = true
        is NewsState.FetchSuccess, is NewsState.FetchFailed -> swipeRefreshLayout.isRefreshing = false
    }
}

这样下拉刷新控件会根据请求状态自动锁定/解锁，防止用户重复触发刷新操作。

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