使用Ktor下载文件并通过Flow/CallbackFlow展示进度的问题

相关实现代码 #

追踪下载状态的密封类 #

sealed class DownloadType {
    object Started : DownloadType()

    class Progress(
        val value: String
    ) : DownloadType()

    class Finished(
        val file: File
    ) : DownloadType()

    class Failed(
        val throwable: Throwable
    ) : DownloadType()
}

Ktor的Hilt依赖注入模块 #

@Module
@InstallIn(SingletonComponent::class)
object KtorModule {
    @Provides
    @Singleton
    fun provideHttpClient(): HttpClient = HttpClient(Android)
}

下载用例接口 #

interface DownloadFileUseCase {
    suspend operator fun invoke(
        url: String,
        name: String
    ): Flow<DownloadType>
}

基于普通Flow的实现（存在异常问题） #

class DownloadFileUseCaseImplFlow @Inject constructor(
    @ApplicationContext private val context: Context,
    private val httpClient: HttpClient
) : DownloadFileUseCase {
    override suspend fun invoke(
        url: String,
        name: String
    ): Flow<DownloadType> = flow {
        emit(DownloadType.Started)

        val path = File("${context.filesDir}/example")
        if (!path.exists()) path.mkdirs()

        val bytesChannel = httpClient.get(url) {
            onDownload { bytesSentTotal, contentLength ->
                // 此处直接emit会触发异常，无法正常运行
                emit(DownloadType.Progress(value = "$bytesSentTotal of $contentLength"))
            }
        }.bodyAsChannel()

        val file = File(path, name)
        bytesChannel.copyAndClose(file.writeChannel())

        emit(DownloadType.Finished(file = file))
    }.catch { error ->
        error.printStackTrace()
        emit(DownloadType.Failed(throwable = error))
    }
}

基于callbackFlow的实现（遇到速度问题） #

class DownloadFileUseCaseImplCallbackFlow @Inject constructor(
    @ApplicationContext private val context: Context,
    private val httpClient: HttpClient
) : DownloadFileUseCase {
    override suspend fun invoke(
        url: String,
        name: String
    ): Flow<DownloadType> = callbackFlow {
        send(DownloadType.Started)

        val path = File("${context.filesDir}/example")
        if (!path.exists()) path.mkdirs()

        val bytesChannel = httpClient.get(url) {
            onDownload { bytesSentTotal, contentLength ->
                // 场景1：频繁发送进度，下载速度显著变慢
                // send(DownloadType.Progress(value = "$bytesSentTotal - $contentLength"))
                // 场景2：仅发送一次进度提示，下载速度正常
                // send(DownloadType.Progress(value = "Downloading file size $contentLength, please wait..."))
            }
        }.bodyAsChannel()

        val file = File(path, name)
        bytesChannel.copyAndClose(file.writeChannel())

        send(DownloadType.Finished(file = file))

        close()
    }.catch { error ->
        error.printStackTrace()
        emit(DownloadType.Failed(throwable = error))
    }
}

遇到的问题 #

使用同一文件测试时，场景1在onDownload回调中针对每个bytesSentTotal调用send发送进度，下载速度明显变慢；场景2仅调用一次send，下载速度恢复正常。两种场景下，ViewModel均使用collectLatest收集Flow。

原因分析 #

1. callbackFlow的send特性：send是挂起函数，必须等待下游收集器处理完当前元素后才会继续执行。Ktor的onDownload会频繁触发（每接收一小段数据就调用一次），每次调用都触发send会导致下载线程被频繁挂起，等待下游处理进度更新，挤占了下载数据的时间，最终拖慢整体速度。
2. collectLatest的额外开销：虽然collectLatest会取消前一次未完成的收集任务，但取消操作本身存在开销，叠加频繁send导致的挂起等待，进一步加剧了下载效率的下降。

解决方案 #

核心思路是限制进度发送的频率，避免过于频繁的send操作，同时保证进度更新的实时性。

方案1：定时节流发送进度（推荐） #

在callbackFlow中启动协程，定时发送最新的进度数据，比如每500ms更新一次：

override suspend fun invoke(
    url: String,
    name: String
): Flow<DownloadType> = callbackFlow {
    send(DownloadType.Started)

    val path = File("${context.filesDir}/example")
    if (!path.exists()) path.mkdirs()

    var latestBytesSent = 0L
    var totalContentLength = 0L

    val bytesChannel = httpClient.get(url) {
        onDownload { bytesSentTotal, contentLength ->
            // 仅更新进度变量，不直接发送
            latestBytesSent = bytesSentTotal
            totalContentLength = contentLength
        }
    }.bodyAsChannel()

    // 启动定时发送协程
    launch {
        while (isActive) {
            send(DownloadType.Progress(value = "$latestBytesSent of $totalContentLength"))
            delay(500) // 每500ms发送一次最新进度
        }
    }

    val file = File(path, name)
    bytesChannel.copyAndClose(file.writeChannel())

    send(DownloadType.Finished(file = file))
    close()
}.catch { error ->
    error.printStackTrace()
    emit(DownloadType.Failed(throwable = error))
}

方案2：按进度阈值发送 #

设置进度变化阈值，比如每下载1MB或文件的1%才发送一次进度，减少send次数：

override suspend fun invoke(
    url: String,
    name: String
): Flow<DownloadType> = callbackFlow {
    send(DownloadType.Started)

    val path = File("${context.filesDir}/example")
    if (!path.exists()) path.mkdirs()

    var lastReportedBytes = 0L
    val progressThreshold = 1024 * 1024 // 每1MB更新一次进度

    val bytesChannel = httpClient.get(url) {
        onDownload { bytesSentTotal, contentLength ->
            // 达到阈值或下载完成时才发送进度
            if (bytesSentTotal - lastReportedBytes >= progressThreshold || bytesSentTotal == contentLength) {
                send(DownloadType.Progress(value = "$bytesSentTotal of $contentLength"))
                lastReportedBytes = bytesSentTotal
            }
        }
    }.bodyAsChannel()

    val file = File(path, name)
    bytesChannel.copyAndClose(file.writeChannel())

    send(DownloadType.Finished(file = file))
    close()
}.catch { error ->
    error.printStackTrace()
    emit(DownloadType.Failed(throwable = error))
}

方案3：使用offer替代send #

offer是非挂起函数，尝试发送元素时不会等待下游处理，队列满时会丢弃旧元素。可配合buffer调整缓冲区大小，避免频繁丢弃：

override suspend fun invoke(
    url: String,
    name: String
): Flow<DownloadType> = callbackFlow {
    send(DownloadType.Started)

    val path = File("${context.filesDir}/example")
    if (!path.exists()) path.mkdirs()

    val bytesChannel = httpClient.get(url) {
        onDownload { bytesSentTotal, contentLength ->
            // 使用offer避免挂起等待
            offer(DownloadType.Progress(value = "$bytesSentTotal of $contentLength"))
        }
    }.bodyAsChannel()

    val file = File(path, name)
    bytesChannel.copyAndClose(file.writeChannel())

    send(DownloadType.Finished(file = file))
    close()
}.buffer(Channel.UNLIMITED) // 调整缓冲区大小，减少进度丢失
.catch { error ->
    error.printStackTrace()
    emit(DownloadType.Failed(throwable = error))
}

补充说明 #

• 方案1的定时节流平衡了实时性和性能，适合大多数UI场景；
• 方案2适合大文件下载，减少不必要的更新；
• 方案3的offer方式不会阻塞下载线程，但可能丢失部分进度更新，适合对进度精度要求不高的场景。

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