Kotlin协程如何实现单协程读文件、另一协程经Channel逐行打印的规范写法

你遇到的编译错误本质不是协程的实现层级问题，是Kotlin的类型安全校验导致的：

• File.forEachLine 是JDK提供的同步IO工具方法，它接收的行处理lambda是普通的 (String) -> Unit 类型，没有suspend修饰，类型层面就不允许在内部调用send这类挂起函数，和Java不允许在普通Runnable中抛出checked异常是同一类设计逻辑。
• 就算绕开编译校验硬写出来，原代码还有一个隐藏的逻辑漏洞：Channel全程没有主动关闭，接收端的for (msg in ch)循环会永远挂起等待新内容，导致整个程序永远阻塞无法退出。

核心原则是 不要把挂起函数塞进非挂起的回调lambda里，把逐行读取文件的整个流程放到协程的挂起上下文中执行：

fun main() = runBlocking {
    val ch = Channel<String>()
    // 接收端协程：逐行消费内容
    launch {
        for (msg in ch) {
            println("行长度：${msg.length}, 内容：$msg")
        }
        println("所有内容处理完成")
    }
    // 发送端协程：逐行读取文件发送
    launch {
        // 用bufferedReader按行读，整个读流程都处于协程挂起上下文
        File("file.txt").bufferedReader().use { reader ->
            var line: String? = reader.readLine()
            while (line != null) {
                ch.send(line) // 这里可以正常调用挂起函数send
                line = reader.readLine()
            }
        }
        // 读完文件必须主动关闭Channel，否则接收端会永久挂起
        ch.close()
    }
}

注意事项：

• 不要为了适配forEachLine去硬套runBlocking包在send外面，那样会完全抵消协程的非阻塞优势，和普通同步代码没有区别。
• 默认创建的Channel是无缓冲类型，send操作会挂起直到接收端取走对应内容，行为和Golang的无缓冲channel完全一致；如果需要缓冲能力，可以在构造Channel、或者调用produce时传入capacity参数指定缓冲区大小。

上面手动管理Channel的写法虽然能跑，但很容易漏处理Channel关闭、异常传播、协程取消的边界逻辑，Kotlin官方提供了produce构建器专门做这类生产-消费的场景：

// 把文件逐行读取的逻辑封装成一个自动管理Channel的生产者
fun CoroutineScope.readFileLines(file: File) = produce<String> {
    file.bufferedReader().use { reader ->
        var line: String? = reader.readLine()
        while (line != null) {
            send(line)
            line = reader.readLine()
        }
    }
    // 生产逻辑结束后，produce构建器会自动关闭底层Channel，不需要手动调用close
}

使用的时候直接拿到这个Channel就行：

fun main() = runBlocking {
    val lineChannel = readFileLines(File("file.txt"))
    // 消费端逻辑完全一致
    launch {
        for (msg in lineChannel) {
            println(msg.length)
        }
    }
}

这种写法的好处是当上游生产协程因为异常取消、或者正常执行结束，底层Channel会自动关闭，消费端的循环可以正常退出，不会出现永久挂起的问题。

这个场景和读文件的逻辑没有本质区别，只需要把文件读取的部分换成子进程输入流的读取即可，完全复用上面的produce封装思路：

// 逐行读取子进程的stdout输出，自动管理Channel
fun CoroutineScope.readProcessStdout(process: Process) = produce<String> {
    process.inputStream.bufferedReader().use { reader ->
        var line: String? = reader.readLine()
        while (line != null) {
            // 过滤空行，很多子进程会输出空行做格式分隔
            if (line.isNotBlank()) send(line.trim())
            line = reader.readLine()
        }
    }
    // 等待子进程执行结束，避免资源泄漏
    process.waitFor()
}

// 业务逻辑示例
fun main() = runBlocking {
    // 启动目标子进程
    val targetProcess = ProcessBuilder("your-target-command").start()
    // 拿到逐行输出的Channel
    val eventChannel = readProcessStdout(targetProcess)
    // 开消费协程逐行解析处理JSON
    launch {
        for (rawLine in eventChannel) {
            // 直接在挂起上下文里完成JSON解析、业务请求、后续逻辑处理即可
            val event = jsonDecoder.decodeFromString<BusinessEvent>(rawLine)
            handleBusinessEvent(event)
        }
    }
}

你提到的Golang的写法在Kotlin里不能直接用，本质是两个并发模型的设计目标不同：

• Golang的send操作是直接阻塞当前goroutine，没有上下文校验，所以可以在任意位置调用。
• Kotlin协程通过类型系统强制把挂起函数的调用限定在协程上下文中，本质是帮你提前规避把阻塞IO、重计算切错线程的问题——比如在Android主线程、或者Web服务的请求处理线程里误调用同步文件读、同步进程IO操作，避免线程阻塞导致的性能问题。
• 如果确实需要对接老的、只提供非挂起回调的第三方库，可以用suspendCancellableCoroutine把回调逻辑封装成标准的挂起函数，不要硬在非挂起lambda里强行调用挂起操作。

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