协程的并发控制、限流与节流、批量处理与优先级调度

好，咱们接着聊协程的并发控制。上一章我们打下了基础，这一章要深入一些实战中绕不开的硬骨头。

说白了，协程虽然轻量，但也不是无限资源。你想想看，如果一下子启动几千个协程去抢数据库连接，或者同时发几百个网络请求，系统照样扛不住。我早年在一个社交App项目里就吃过这个亏——用户刷朋友圈时，一个列表项就启动一个协程去加载图片，结果瞬间几百个协程同时跑，内存直接飙到OOM。嗯，从那以后，我对并发控制就格外上心。

一、协程的并发控制

并发控制的核心，就是限制同时运行的协程数量。Kotlin协程库提供了Mutex和Semaphore两种经典工具，但更常用的是limitedParallelism。

核心思路：用有限的线程池或信号量，把并发度控制在合理范围内。

1. 使用 Semaphore 控制并发数

val semaphore = Semaphore(3) // 最多3个协程同时执行

suspend fun limitedTask(id: Int) {
    semaphore.acquire()
    try {
        delay(1000)
        println("任务 $id 完成")
    } finally {
        semaphore.release()
    }
}

// 启动10个协程，但只有3个能同时运行
repeat(10) {
    GlobalScope.launch {
        limitedTask(it)
    }
}

我在项目中遇到过类似场景：批量上传图片时，后端接口只允许同时处理5个请求。用Semaphore控制后，既不会把服务器打挂，又能保持上传效率。

小技巧：记得用use函数或者try-finally释放信号量，否则协程泄漏会让你头疼。

2. 使用 limitedParallelism 限制并行度

val dispatcher = Dispatchers.IO
    .limitedParallelism(4) // 最多4个并行任务

suspend fun processData(data: List<Int>) = withContext(dispatcher) {
    data.map { it * 2 }
}

这个API是Kotlin 1.6引入的，我个人习惯用它替代自定义线程池。它底层用的是CoroutineDispatcher的限流机制，比手动管理信号量更优雅。

注意：limitedParallelism只限制在同一个调度器上的并行度。如果你混合使用多个调度器，总并发数可能超出预期。

二、协程的限流与节流

限流和节流，这两个概念经常被混淆。我简单解释一下：

  • 限流（Throttle）：控制单位时间内的执行次数。比如每秒最多处理10个请求。
  • 节流（Debounce）：延迟执行，直到事件停止触发一段时间。比如用户输入停止300ms后再搜索。

在Android开发中，这两个场景太常见了。我记得有一次做搜索框优化，用户每敲一个字就发一次请求，结果接口被频繁调用，体验极差。后来用节流解决了。

1. 限流实现

class ThrottleFirst(private val windowMs: Long = 1000L) {
    private var lastTime = 0L

    suspend fun <T> execute(block: suspend () -> T): T? {
        val now = System.currentTimeMillis()
        if (now - lastTime < windowMs) return null
        lastTime = now
        return block()
    }
}

// 使用
val throttle = ThrottleFirst(500L)
repeat(10) {
    launch {
        throttle.execute {
            println("执行任务 $it")
        }
    }
}

这个实现很简单，但够用。实际项目中，我更喜欢用Flow的throttleFirst操作符，它更符合响应式编程的思维。

2. 节流实现

fun <T> Flow<T>.debounce(timeoutMs: Long): Flow<T> = flow {
    val job = Job()
    collect { value ->
        job.cancel()
        job.complete()
        launch {
            delay(timeoutMs)
            emit(value)
        }
    }
}

// 使用
searchQueryFlow
    .debounce(300L)
    .collect { query ->
        // 执行搜索
    }

嗯，这里要注意：Flow自带的debounce操作符在协程库1.3.0之后就有了，没必要自己造轮子。我上面写的只是为了展示原理。

避坑指南：我曾经在节流时忘记取消上一个协程，导致多个搜索请求同时发出，结果界面数据乱跳。记住，节流的核心是「取消旧任务，启动新任务」。

三、协程的批量处理

批量处理，说白了就是把多个小任务攒起来，一次性处理。这在数据库写入、日志上报、网络请求合并等场景下特别有用。

我做过一个埋点上报模块：用户操作频繁时，每条埋点都单独上报太浪费资源。于是我把埋点数据先缓存起来，每100条或者每5秒批量上报一次。效果立竿见影，网络请求量减少了80%。

1. 使用 Channel 实现批量收集

suspend fun batchProcess() {
    val channel = Channel<String>(Channel.UNLIMITED)
    
    // 消费者：批量处理
    val consumer = launch {
        val batch = mutableListOf<String>()
        for (item in channel) {
            batch.add(item)
            if (batch.size >= 10) {
                println("批量处理: $batch")
                batch.clear()
            }
        }
    }
    
    // 生产者：发送数据
    repeat(25) {
        channel.send("数据$it")
        delay(100)
    }
    
    channel.close()
    consumer.join()
}

这个模式叫「生产者-消费者」，用Channel做缓冲区，消费者攒够一批就处理。实际项目中，我还会加一个超时机制：如果5秒内没攒够10条，也强制处理一次。

2. 使用 Flow 的 buffer 和 chunked

dataFlow
    .buffer(50) // 缓冲50个元素
    .chunked(10) // 每10个一组
    .collect { batch ->
        // 批量处理
        saveToDatabase(batch)
    }

chunked是Kotlin标准库的扩展函数，配合Flow的buffer，能轻松实现批量处理。我个人习惯用这种方式，代码更简洁。

建议：批量处理时，记得考虑事务一致性。如果一批数据中某条失败，是全部回滚还是跳过？这取决于你的业务场景。

四、协程的优先级调度

协程本身没有优先级概念，但我们可以通过调度器来实现类似的效果。说白了，就是让重要的任务先执行，不重要的任务靠后站。

我记得在做一个视频播放器时，UI渲染协程必须优先于缓存预加载协程。如果两者抢资源，用户会感觉到卡顿。

1. 使用不同的调度器

// 高优先级：UI相关
val uiDispatcher = Dispatchers.Main

// 中优先级：数据处理
val dataDispatcher = Dispatchers.Default

// 低优先级：后台任务
val backgroundDispatcher = Dispatchers.IO.limitedParallelism(2)

GlobalScope.launch(uiDispatcher) {
    // 更新UI
}

GlobalScope.launch(dataDispatcher) {
    // 处理数据
}

GlobalScope.launch(backgroundDispatcher) {
    // 预加载缓存
}

这种方法简单粗暴，但够用。不同调度器底层使用不同的线程池，自然就有了优先级差异。

2. 使用自定义优先级队列

class PriorityCoroutineDispatcher(
    private val dispatcher: CoroutineDispatcher
) : CoroutineDispatcher() {
    
    override fun dispatch(context: CoroutineContext, block: Runnable) {
        val priority = context[PriorityKey]?.priority ?: 0
        // 根据优先级决定是否立即执行
        if (priority > 5) {
            dispatcher.dispatch(context, block)
        } else {
            // 放入延迟队列
            delayQueue.offer(priority, block)
        }
    }
}

object PriorityKey : CoroutineContext.Key<PriorityElement>
data class PriorityElement(val priority: Int) : CoroutineContext.Element {
    override val key: CoroutineContext.Key<*> get() = PriorityKey
}

这个实现比较高级，我一般只在框架层使用。日常开发中，用不同调度器就足够了。

注意：不要滥用优先级。如果所有任务都标为高优先级，那优先级就失去了意义。我见过一个项目，每个协程都用了Dispatchers.Main，结果主线程被塞满，反而更卡。

知识体系总览

下面这张图总结了本章的核心内容，你可以把它当作一个快速参考：

协程并发控制知识体系 并发控制 • Semaphore • limitedParallelism • Mutex • 限制并行度 限流与节流 • Throttle 限流 • Debounce 节流 • Flow 操作符 • 控制执行频率 批量处理 • Channel 收集 • chunked 分组 • buffer 缓冲 • 批量写入/上报 优先级调度 • 不同调度器 • 自定义优先级 • 延迟队列 • 任务分级 实战建议 1. 并发控制：用 limitedParallelism 替代手动信号量 2. 限流节流：优先使用 Flow 内置操作符 3. 批量处理：Channel + 超时机制更灵活 4. 优先级调度：不同调度器比自定义优先级更可靠

这张图把四个核心模块放在一起对比，方便你快速定位。实际开发中，这些技术经常组合使用。比如批量处理时，往往也需要限流控制频率；优先级调度时，可能还要配合并发控制。

好了，这一章的内容就到这里。协程的并发控制、限流节流、批量处理和优先级调度，每个都是实战中的利器。你可以在自己的项目里试着用起来，遇到问题再回来翻翻这篇文章。

公众号:蓝海资料掘金营，微信deep3321
← 上一章 📖 返回目录 下一章 →