如何在Kotlin中将响应式API转换为惰性序列？

完全可以不用后台线程实现需求，Kotlin中惯用的方式是结合协程挂起函数与SequenceScope，或者利用Channel实现拉取驱动的同步，让推送式API的回调与Sequence的迭代在同一线程交替执行。

核心思路 #

推送式API（如Files.walkFileTree）会主动遍历并触发回调，而Sequence是惰性拉取的——只有当迭代器调用next()时才需要下一个元素。通过协程的挂起特性，可以让推送过程在每次触发回调后暂停，直到Sequence请求下一个元素，从而避免额外线程的开销。

实现方案1：基于Channel的Sequence实现 #

利用Kotlin协程的Channel（RENDEZVOUS模式，无缓冲）实现回调与Sequence的同步，整个过程在同一线程执行：

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.SimpleFileVisitor
import java.nio.file.attribute.BasicFileAttributes
import java.util.Optional

typealias Callback<T> = (T) -> Unit

fun <T : Any> sequenceOf(callbackInvoker: (Callback<T>) -> Unit): Sequence<T> = sequence {
    val channel = Channel<Optional<T>>(Channel.RENDEZVOUS)

    // 用Unconfined调度器，协程在当前线程执行，直到挂起点
    val job = GlobalScope.launch(Dispatchers.Unconfined) {
        try {
            callbackInvoker { item ->
                // 回调中发送元素，挂起直到Sequence接收
                runBlocking { channel.send(Optional.of(item)) }
            }
            // 发送结束标记
            channel.send(Optional.empty())
        } finally {
            channel.close()
        }
    }

    try {
        do {
            val item = channel.receive()
            if (item.isPresent) {
                yield(item.get())
            }
        } while (item.isPresent)
    } finally {
        // 终止协程，避免资源泄漏
        job.cancel()
    }
}

// 调用示例
fun main() {
    sequenceOf { onFile: (Path) -> Unit ->
        Files.walkFileTree(Path("."), object : SimpleFileVisitor<Path>() {
            override fun visitFile(file: Path, attrs: BasicFileAttributes): FileVisitResult {
                onFile(file)
                return FileVisitResult.CONTINUE
            }
        })
    }.forEach(::println)
}

实现方案2：更惯用的Flow替代方案 #

如果场景允许，Kotlin的Flow是处理响应式流更惯用的选择，它天生支持惰性与挂起函数，无需额外线程即可实现推送转拉取：

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.SimpleFileVisitor
import java.nio.file.attribute.BasicFileAttributes

fun walkFilesAsFlow(path: Path): Flow<Path> = flow {
    Files.walkFileTree(path, object : SimpleFileVisitor<Path>() {
        override fun visitFile(file: Path, attrs: BasicFileAttributes): FileVisitResult {
            // 用runBlocking适配非挂起回调，发送元素到Flow
            runBlocking { emit(file) }
            return FileVisitResult.CONTINUE
        }
    })
}

// 调用示例
fun main() = runBlocking {
    walkFilesAsFlow(Path(".")).collect(::println)

    // 如果需要Sequence，可以将Flow转换为Sequence
    walkFilesAsFlow(Path(".")).asSequence().forEach(::println)
}

方案优势 #

• 无额外后台线程开销：利用协程挂起特性，让回调执行与Sequence迭代在同一线程交替进行。
• 惰性执行：仅在Sequence请求元素时才处理下一个推送项，符合Sequence的设计初衷。
• 资源安全：通过finally块终止协程/关闭Channel，避免资源泄漏。

内容的提问来源于stack exchange，提问作者Андрей Щеглов