Android 异步编程中协程的完整实战示例

一、全链路数据加载：网络请求 + 数据库缓存

在实际开发中，数据加载通常需要先检查本地缓存，若缓存失效则从网络获取，并将结果更新到本地。以下是完整的 MVVM 架构示例：

1. 项目结构

app/
├── data/               # 数据层
│   ├── model/          # 数据模型
│   │   └── User.kt
│   ├── remote/         # 网络层
│   │   └── UserApiService.kt
│   ├── local/          # 本地数据库（Room）
│   │   ├── UserDao.kt
│   │   └── AppDatabase.kt
│   └── repository/     # 仓库层
│       └── UserRepository.kt
├── ui/                 # UI 层
│   └── UserListActivity.kt
└── viewmodel/          # ViewModel 层└── UserListViewModel.kt

2. 关键代码实现

2.1 数据模型（User.kt）

@Entity(tableName = "users")
data class User(@PrimaryKey val id: String,val name: String,val age: Int,val lastUpdateTime: Long = System.currentTimeMillis() // 缓存时间戳
)

2.2 网络层（UserApiService.kt）

使用 Retrofit 定义挂起函数（协程友好）：

interface UserApiService {@GET("users")suspend fun getUsersFromNetwork(): Response<List<User>>
}

2.3 本地数据库（UserDao.kt）

Room DAO 支持协程（suspend 函数自动在 IO 线程执行）：

@Dao
interface UserDao {@Query("SELECT * FROM users")suspend fun getCachedUsers(): List<User>@Insert(onConflict = OnConflictStrategy.REPLACE)suspend fun insertUsers(users: List<User>)@Query("DELETE FROM users")suspend fun clearCache()
}

2.4 仓库层（UserRepository.kt）

协程的核心逻辑层，处理网络请求、缓存策略和数据合并：

class UserRepository(private val apiService: UserApiService,private val userDao: UserDao
) {// 缓存有效期（假设 5 分钟）private val CACHE_DURATION = 5 * 60 * 1000L// 获取用户数据（优先缓存，缓存过期则从网络加载）suspend fun getUsers(): Result<List<User>> = withContext(Dispatchers.IO) {try {// 步骤1：检查本地缓存是否有效val cachedUsers = userDao.getCachedUsers()if (cachedUsers.isNotEmpty() && isCacheValid(cachedUsers)) {return@withContext Result.success(cachedUsers)}// 步骤2：缓存无效，从网络获取val response = apiService.getUsersFromNetwork()if (response.isSuccessful) {val remoteUsers = response.body() ?: emptyList()// 步骤3：更新本地缓存userDao.clearCache()userDao.insertUsers(remoteUsers)return@withContext Result.success(remoteUsers)}// 网络请求失败时，返回缓存（即使过期）if (cachedUsers.isNotEmpty()) {return@withContext Result.success(cachedUsers)}Result.failure(Exception("网络请求失败且无缓存"))} catch (e: Exception) {Result.failure(e)}}// 检查缓存是否有效（取最新一条数据的时间戳）private fun isCacheValid(users: List<User>): Boolean {val latestTime = users.maxOfOrNull { it.lastUpdateTime } ?: 0Lreturn System.currentTimeMillis() - latestTime < CACHE_DURATION}
}

2.5 ViewModel 层（UserListViewModel.kt）

使用 viewModelScope 启动协程，管理数据加载状态：

class UserListViewModel(private val repository: UserRepository
) : ViewModel() {private val _uiState = MutableStateFlow<UiState>(UiState.Loading)val uiState: StateFlow<UiState> = _uiState.asStateFlow()fun loadUsers() {viewModelScope.launch {_uiState.value = UiState.Loadingwhen (val result = repository.getUsers()) {is Result.Success -> {_uiState.value = UiState.Success(result.data)}is Result.Failure -> {_uiState.value = UiState.Error(result.exception.message)}}}}sealed class UiState {object Loading : UiState()data class Success(val users: List<User>) : UiState()data class Error(val message: String?) : UiState()}class UserListViewModelFactory(private val repository: UserRepository) : ViewModelProvider.Factory {override fun <T : ViewModel> create(modelClass: Class<T>): T {if (modelClass.isAssignableFrom(UserListViewModel::class.java)) {@Suppress("UNCHECKED_CAST")return UserListViewModel(repository) as T}throw IllegalArgumentException("Unknown ViewModel class")}}
}

2.6 UI 层（UserListActivity.kt）

观察 StateFlow 并更新 UI：

class UserListActivity : AppCompatActivity() {private lateinit var binding: ActivityUserListBindingprivate val viewModel: UserListViewModel by viewModels {UserListViewModelFactory(UserRepository()) // 这里需要提供UserRepository的实例}override fun onCreate(savedInstanceState: Bundle?) {super.onCreate(savedInstanceState)binding = ActivityUserListBinding.inflate(layoutInflater)setContentView(binding.root)// 观察 UI 状态lifecycleScope.launch {viewModel.uiState.collect { state ->when (state) {is UserListViewModel.UiState.Loading -> showLoading()is UserListViewModel.UiState.Success -> showUsers(state.users)is UserListViewModel.UiState.Error -> showError(state.message)}}}// 触发数据加载viewModel.loadUsers()}private fun showLoading() {binding.progressBar.visibility = View.VISIBLEbinding.recyclerView.visibility = View.GONEbinding.errorText.visibility = View.GONE}private fun showUsers(users: List<User>) {binding.progressBar.visibility = View.GONEbinding.recyclerView.visibility = View.VISIBLEbinding.errorText.visibility = View.GONE// 初始化 RecyclerView 并设置适配器binding.recyclerView.adapter = UserAdapter(users)}private fun showError(message: String?) {binding.progressBar.visibility = View.GONEbinding.recyclerView.visibility = View.GONEbinding.errorText.visibility = View.VISIBLEbinding.errorText.text = message ?: "加载失败"}
}

二、协程的取消与资源清理

在协程中执行文件操作、网络请求或打开数据库连接时，需要确保协程取消时释放资源。以下是资源清理的完整示例：

2.1 取消协程时关闭文件

// 在 ViewModel 中启动一个协程，读取大文件并处理
fun processLargeFile(filePath: String) {viewModelScope.launch {val file = File(filePath)val inputStream = file.inputStream()try {// 模拟逐行读取文件（可取消）var line: String?while (isActive) { // 检查协程是否活跃line = inputStream.bufferedReader().readLine()if (line == null) breakprocessLine(line) // 处理每一行数据}} finally {// 协程取消时，确保关闭文件流inputStream.close()Log.d("FileProcess", "文件流已关闭")}}
}

2.2 取消网络请求（Retrofit + 协程）

Retrofit 的 Call 对象支持协程取消，协程取消时会自动取消底层的网络请求：

// 定义可取消的网络请求
suspend fun fetchData(): Result<Data> = withContext(Dispatchers.IO) {try {val response = apiService.getData() // Retrofit 的 suspend 函数if (response.isSuccessful) {Result.success(response.body()!!)} else {Result.failure(Exception("HTTP 错误: ${response.code()}"))}} catch (e: CancellationException) {// 协程被取消时触发，可在此记录日志或清理资源Log.d("Network", "请求被取消")throw e // 重新抛出，确保上层知道协程已取消} catch (e: Exception) {Result.failure(e)}
}

三、Flow 的高级用法：处理背压与热流

3.1 背压（Backpressure）处理

当生产者发射数据过快，消费者处理不过来时，使用 conflate（取最新值）或 buffer（缓存数据）解决背压问题：

// 模拟传感器数据（每秒发射 100 次）
fun sensorDataFlow(): Flow<Int> = flow {var value = 0while (true) {emit(value++)delay(10) // 10ms 发射一次（100Hz）}
}.flowOn(Dispatchers.IO)// 在 ViewModel 中收集数据（每秒处理 1 次）
fun startSensorMonitoring() {viewModelScope.launch {sensorDataFlow().conflate() // 只处理最新值，丢弃中间未处理的数据// .buffer(10) // 缓存 10 个数据，超出则挂起生产者.collect { value ->delay(1000) // 模拟耗时处理（1Hz）_sensorValue.value = value}}
}

3.2 SharedFlow：多订阅者热流

SharedFlow 适用于多个订阅者需要接收同一数据流的场景（如事件广播）：

// 在 Repository 中定义 SharedFlow
class EventRepository {private val _eventFlow = MutableSharedFlow<Event>()val eventFlow: SharedFlow<Event> = _eventFlow.asSharedFlow()// 发送事件（如网络状态变化）suspend fun sendEvent(event: Event) {_eventFlow.emit(event)}
}// 在多个 Activity/Fragment 中订阅
lifecycleScope.launch {eventRepository.eventFlow.collect { event ->when (event) {is Event.NetworkConnected -> updateNetworkStatus(true)is Event.NetworkDisconnected -> updateNetworkStatus(false)}}
}

四、协程与 WorkManager 集成：后台任务

WorkManager 是 Android 官方的后台任务调度库，支持协程。以下是使用协程实现后台数据同步的示例：

4.1 定义协程 Worker

class DataSyncWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) {override suspend fun doWork(): Result {return withContext(Dispatchers.IO) {try {// 执行后台数据同步（调用 Repository）val result = repository.syncData()if (result.isSuccess) {Result.success()} else {Result.retry() // 失败后重试}} catch (e: Exception) {Result.failure()}}}
}

4.2 调度后台任务

// 在需要的地方（如 Application）调度每日同步
val workRequest = PeriodicWorkRequestBuilder<DataSyncWorker>(1, TimeUnit.DAYS).setConstraints(Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build()).build()WorkManager.getInstance(context).enqueue(workRequest)

五、协程测试：使用 TestCoroutineDispatcher

测试协程代码时，需控制协程的执行时间和顺序。使用 kotlinx-coroutines-test 库中的 TestCoroutineDispatcher 或 runTest 方法：

5.1 单元测试示例

class UserRepositoryTest {private lateinit var repository: UserRepositoryprivate lateinit var testDispatcher: TestDispatcher@Beforefun setup() {testDispatcher = UnconfinedTestDispatcher() // 无限制调度器（立即执行）val apiService = mockk<UserApiService>()val userDao = mockk<UserDao>()repository = UserRepository(apiService, userDao)}@Testfun `getUsers 缓存有效时返回缓存数据`() = runTest(testDispatcher) {// 模拟缓存数据（有效期内）val cachedUsers = listOf(User("1", "Alice", 20, System.currentTimeMillis()))every { userDao.getCachedUsers() } returns cachedUsersval result = repository.getUsers()assertTrue(result is Result.Success)assertEquals(cachedUsers, (result as Result.Success).data)}@Testfun `getUsers 缓存过期时从网络加载`() = runTest(testDispatcher) {// 模拟过期缓存val expiredUsers = listOf(User("1", "Alice", 20, System.currentTimeMillis() - 10 * 60 * 1000))every { userDao.getCachedUsers() } returns expiredUsers// 模拟网络成功响应val remoteUsers = listOf(User("2", "Bob", 25))coEvery { apiService.getUsersFromNetwork() } returns Response.success(remoteUsers)val result = repository.getUsers()// 验证网络请求被调用，且缓存被更新coVerify { apiService.getUsersFromNetwork() }coVerify { userDao.insertUsers(remoteUsers) }assertTrue(result is Result.Success)assertEquals(remoteUsers, (result as Result.Success).data)}
}

六、总结：协程的完整使用规范

通过以上示例，可以总结出 Android 协程开发的最佳实践：

1. 结构化并发：始终使用 lifecycleScope 或 viewModelScope 管理协程生命周期，避免内存泄漏。
2. 明确线程分工：IO 操作使用 Dispatchers.IO，计算任务使用 Dispatchers.Default，UI 更新使用 Dispatchers.Main（默认）。
3. 异常处理分层：
   • 网络 / 数据库层：返回 Result 类型或抛出可恢复异常。
   • ViewModel 层：统一捕获异常并转换为 UI 状态（如 Loading、Error）。
   • UI 层：根据状态更新界面，避免在协程内直接操作 UI（通过 StateFlow/LiveData 间接更新）。
4. 资源清理：使用 try-finally 或 use 方法确保文件流、网络连接等资源在协程取消时释放。
5. 测试覆盖：使用 runTest 和 TestDispatcher 测试协程逻辑，验证数据加载、缓存策略和异常处理的正确性。

通过遵循这些规范，协程能显著提升 Android 异步代码的可读性、可维护性和健壮性，是现代 Android 开发的核心工具之一。