当前位置：首页 > news >正文

Kotlin实现文件上传进度监听：RequestBody封装详解

news 来源：原创 2025/6/16 12:50:12

引言：为什么需要文件上传进度监听？

在移动应用开发中，文件上传是常见需求。特别是当用户上传大文件（如视频、高清图片）时，缺乏进度反馈会导致糟糕的用户体验。本文将深入探讨如何通过封装OkHttp的RequestBody实现高效、灵活的文件上传进度监听。

技术架构图

关键技术组件

ProgressListener：进度回调接口
CountingSink：字节计数拦截器
ProgressRequestBody：RequestBody封装器

完整代码实现

1. 进度监听接口

interface ProgressListener {fun onProgress(bytesWritten: Long, contentLength: Long, identifier: String? = null,done: Boolean = false)
}

2. 字节计数Sink实现

import okio.ForwardingSink
import okio.Sink
import java.io.IOExceptionclass CountingSink(delegate: Sink,private val listener: ProgressListener,private val contentLength: Long,private val identifier: String? = null
) : ForwardingSink(delegate) {private var bytesWritten = 0Lprivate var lastReportedPercent = -1@Throws(IOException::class)override fun write(source: Buffer, byteCount: Long) {super.write(source, byteCount)bytesWritten += byteCount// 优化：每1%进度回调一次，避免频繁更新UIval currentPercent = (100 * bytesWritten / contentLength).toInt()if (currentPercent != lastReportedPercent) {listener.onProgress(bytesWritten, contentLength, identifier)lastReportedPercent = currentPercent}}@Throws(IOException::class)override fun close() {super.close()// 最终完成回调listener.onProgress(bytesWritten, contentLength, identifier, true)}
}

3. 可监听进度的RequestBody封装

import okhttp3.MediaType
import okhttp3.RequestBody
import okio.BufferedSink
import okio.Okioclass ProgressRequestBody(private val delegate: RequestBody,private val listener: ProgressListener,private val identifier: String? = null
) : RequestBody() {override fun contentType(): MediaType? = delegate.contentType()override fun contentLength(): Long = delegate.contentLength()@Throws(IOException::class)override fun writeTo(sink: BufferedSink) {val countingSink = CountingSink(sink, listener, contentLength(),identifier)val bufferedSink = Okio.buffer(countingSink)delegate.writeTo(bufferedSink)// 确保所有数据写入bufferedSink.flush()}
}

实战应用：多文件上传示例

Retrofit服务接口定义

interface UploadService {@Multipart@POST("upload")suspend fun uploadFiles(@Part files: List<MultipartBody.Part>): Response<UploadResult>
}

多文件上传管理器

import android.os.Handler
import android.os.Looper
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import okhttp3.MultipartBody
import retrofit2.Response
import java.io.Fileclass UploadManager(private val service: UploadService,private val uiHandler: Handler = Handler(Looper.getMainLooper())
) {// 进度回调映射：文件路径 -> 进度百分比private val progressMap = mutableMapOf<String, Int>()// 进度监听器实现private val progressListener = object : ProgressListener {override fun onProgress(bytesWritten: Long, contentLength: Long, identifier: String?, done: Boolean) {identifier?.let { filePath ->val progress = (100 * bytesWritten / contentLength).toInt()progressMap[filePath] = progress// 更新UI（切换到主线程）uiHandler.post {// 这里简化处理，实际应通过LiveData或回调通知println("文件 $filePath 上传进度: $progress%")}}}}suspend fun uploadFiles(files: List<File>): UploadResult {return withContext(Dispatchers.IO) {// 准备上传部件val parts = files.map { file ->val requestBody = file.asRequestBody("application/octet-stream".toMediaType())val progressBody = ProgressRequestBody(requestBody,progressListener,file.absolutePath // 使用文件路径作为标识符)MultipartBody.Part.createFormData("files", file.name, progressBody)}// 执行上传val response = service.uploadFiles(parts)if (response.isSuccessful) {response.body() ?: throw UploadException("上传成功但返回空数据")} else {throw UploadException("上传失败: ${response.code()}")}}}
}class UploadException(message: String) : Exception(message)

Android中使用示例

// 在ViewModel中
class UploadViewModel : ViewModel() {private val uploadManager = UploadManager(RetrofitClient.uploadService)// 使用LiveData跟踪进度和结果val uploadProgress = MutableLiveData<Map<String, Int>>()val uploadResult = MutableLiveData<Result<UploadResult>>()fun uploadFiles(files: List<File>) {viewModelScope.launch {uploadResult.value = Result.loading()try {val result = uploadManager.uploadFiles(files)uploadResult.value = Result.success(result)} catch (e: Exception) {uploadResult.value = Result.error(e)}}}// 更新进度（实际应用中UploadManager应回调此方法）fun updateProgress(progressMap: Map<String, Int>) {uploadProgress.value = progressMap}
}// 在Activity/Fragment中观察
viewModel.uploadProgress.observe(this) { progressMap ->progressMap.forEach { (filePath, progress) ->// 更新对应文件的进度条fileProgressBars[filePath]?.progress = progress}
}viewModel.uploadResult.observe(this) { result ->when (result.status) {Status.LOADING -> showLoading()Status.SUCCESS -> showSuccess(result.data)Status.ERROR -> showError(result.message)}
}

性能优化策略

1. 回调频率控制

// 在CountingSink中添加优化逻辑
private var lastReportedTime = 0L
private val REPORT_INTERVAL = 200L // 200毫秒override fun write(source: Buffer, byteCount: Long) {super.write(source, byteCount)bytesWritten += byteCountval currentTime = System.currentTimeMillis()if (currentTime - lastReportedTime > REPORT_INTERVAL || bytesWritten == contentLength) {listener.onProgress(bytesWritten, contentLength, identifier)lastReportedTime = currentTime}
}

2. 弱引用防止内存泄漏

class WeakProgressListener(private val delegate: ProgressListener
) : ProgressListener {private val weakRef = WeakReference(delegate)override fun onProgress(bytesWritten: Long, contentLength: Long, identifier: String?, done: Boolean) {weakRef.get()?.onProgress(bytesWritten, contentLength, identifier, done)}
}// 使用方式
val progressBody = ProgressRequestBody(requestBody,WeakProgressListener(progressListener),file.absolutePath
)

3. 大文件分块上传

class ChunkedUploader(private val file: File,private val chunkSize: Long = 1024 * 1024 // 1MB
) {suspend fun uploadWithProgress(listener: ProgressListener) {val totalSize = file.length()var uploaded = 0Lvar chunkIndex = 0file.inputStream().use { input ->val buffer = ByteArray(chunkSize.toInt())var bytesRead: Intwhile (input.read(buffer).also { bytesRead = it } != -1) {// 上传当前分块uploadChunk(chunkIndex, buffer, bytesRead)// 更新进度uploaded += bytesReadlistener.onProgress(uploaded, totalSize, file.absolutePath)chunkIndex++}}// 最终完成回调listener.onProgress(totalSize, totalSize, file.absolutePath, true)}private suspend fun uploadChunk(index: Int, data: ByteArray, size: Int) {// 实现分块上传逻辑}
}

与其他技术对比

技术方案	优点	缺点	适用场景
RequestBody封装	底层实现、高效灵活、与OkHttp无缝集成	需要自定义封装、对新手有一定难度	需要精细控制上传过程的场景
Interceptor拦截器	统一处理所有请求、位置集中	难以区分不同文件进度、实现复杂	需要全局监控的场景
系统级进度监听	简单易用、无需额外代码	功能有限、无法自定义回调频率	简单小文件上传
分块上传+自定义协议	支持断点续传、精确控制	实现复杂、需要服务端配合	超大文件上传、不稳定网络环境

关键点总结

核心原理：通过自定义Sink拦截写入操作实现字节计数
进度计算：进度百分比 = (已上传字节数 / 文件总大小) * 100
线程安全：进度回调在IO线程，更新UI需切主线程
性能优化：
- 控制回调频率（时间阈值或进度阈值）
- 使用弱引用防止内存泄漏
- 大文件采用分块上传策略
多文件支持：为每个文件分配唯一标识符
容错处理：处理除零异常和取消上传逻辑

高级应用场景

1. 断点续传实现

class ResumableProgressRequestBody(delegate: RequestBody,listener: ProgressListener,identifier: String?,private val startPosition: Long // 从上次中断处继续
) : ProgressRequestBody(delegate, listener, identifier) {override fun writeTo(sink: BufferedSink) {val countingSink = CountingSink(sink, listener, contentLength(),identifier).apply {bytesWritten = startPosition // 设置起始位置}val bufferedSink = Okio.buffer(countingSink)delegate.writeTo(bufferedSink)bufferedSink.flush()}
}

2. 上传速度计算

// 在ProgressListener中添加速度回调
interface AdvancedProgressListener : ProgressListener {fun onSpeedCalculated(bytesPerSecond: Double, identifier: String?)
}// 在CountingSink中实现速度计算
private var lastBytesWritten = 0L
private var lastTimeMillis = System.currentTimeMillis()override fun write(source: Buffer, byteCount: Long) {super.write(source, byteCount)val currentTime = System.currentTimeMillis()val elapsed = currentTime - lastTimeMillisif (elapsed > 1000) { // 每秒计算一次val deltaBytes = bytesWritten - lastBytesWrittenval speed = deltaBytes / (elapsed / 1000.0)(listener as? AdvancedProgressListener)?.onSpeedCalculated(speed, identifier)lastBytesWritten = bytesWrittenlastTimeMillis = currentTime}
}

最佳实践建议

进度显示策略：
- 小文件（<5MB）：显示百分比
- 中等文件（5-50MB）：显示百分比+剩余时间
- 大文件（>50MB）：显示百分比+速度+剩余时间

异常处理增强：

override fun writeTo(sink: BufferedSink) {try {// ... 正常写入逻辑} catch (e: IOException) {listener.onError(e, identifier)throw e} finally {// 清理资源}
}

取消上传支持：

class CancellableProgressRequestBody(delegate: RequestBody,listener: ProgressListener,identifier: String?,private val isCancelled: () -> Boolean // 取消状态检查
) : ProgressRequestBody(delegate, listener, identifier) {@Throws(IOException::class)override fun writeTo(sink: BufferedSink) {val countingSink = object : CountingSink(sink, listener, contentLength(), identifier) {override fun write(source: Buffer, byteCount: Long) {if (isCancelled()) throw IOException("Upload cancelled")super.write(source, byteCount)}}// ... 其余逻辑}
}