WebFlux中执行IO操作的后端为何导致前端访问阻塞？

核心问题根源 #

WebFlux基于Reactor的非阻塞模型，核心要求是绝对不能在EventLoop线程上执行阻塞IO操作。你的几段代码出现差异的本质，就是是否遵守了这个规则：

1. 第一段代码阻塞的原因 #

这段代码存在两个致命问题：

• 阻塞IO占用EventLoop：InputStream.read()是典型的阻塞式IO操作，即使放在suspend fun和coroutineScope中，默认还是运行在WebFlux的EventLoop线程上，直接卡住线程导致请求无法响应。
• 未正确写入响应：手动创建DataBuffer后，没有将其发送到响应流中，前端永远收不到数据，自然一直处于阻塞等待状态。

2. 第二段ZeroCopy代码正常的原因 #

ZeroCopyHttpOutputMessage.writeWith()是操作系统级别的零拷贝操作：数据直接从内核文件缓冲区发送到网卡，完全不经过用户态线程，既不会阻塞EventLoop，又能高效完成文件传输。同时返回的Mono<Void>符合WebFlux的响应式规范，框架能正确处理请求的完成信号。

3. 第三段mono { ... }仍阻塞的原因 #

mono { ... }默认使用Reactor的EventLoop线程执行闭包内的代码，如果里面的IO操作是阻塞的，依然会卡住EventLoop。你需要显式指定将阻塞操作放到专门的阻塞线程池中执行。

修正方案 #

方案1：用WebFlux原生工具处理文件下载（推荐）

直接使用DataBufferUtils实现非阻塞文件读取，完全符合WebFlux规范：

@GetMapping("/download/test")
fun downloadFile(response: ServerHttpResponse): Flux<DataBuffer> {
    val file = File("/home/steiner/Downloads/hello.jpg")
    response.headers.apply {
        contentType = MediaType.APPLICATION_OCTET_STREAM
        contentLength = file.length()
        set(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=hello.jpg")
    }

    return DataBufferUtils.read(file.toPath(), response.bufferFactory(), 8192)
}

方案2：修正协程中的阻塞IO操作

如果必须手动处理流，要将阻塞IO切换到Dispatchers.IO调度器（Kotlin协程的阻塞IO专用线程池）：

@GetMapping("/download/test")
suspend fun downloadFile(response: ServerHttpResponse): Flux<DataBuffer> {
    val file = File("/home/steiner/Downloads/hello.jpg")
    response.headers.apply {
        contentType = MediaType.APPLICATION_OCTET_STREAM
        contentLength = file.length()
        set(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=hello.jpg")
    }

    return Flux.using(
        { file.inputStream() },
        { inputStream ->
            Flux.generate { sink ->
                val buffer = ByteArray(8192)
                val bytesRead = runCatching {
                    // 将阻塞读操作切换到Dispatchers.IO
                    withContext(Dispatchers.IO) {
                        inputStream.read(buffer)
                    }
                }.getOrElse {
                    sink.error(it)
                    return@generate
                }
                if (bytesRead == -1) {
                    sink.complete()
                } else {
                    sink.next(response.bufferFactory().wrap(buffer, 0, bytesRead))
                }
            }
        },
        { inputStream ->
            runCatching {
                withContext(Dispatchers.IO) {
                    inputStream.close()
                }
            }
        }
    )
}

方案3：修正mono { ... }的阻塞问题

显式指定协程调度器为Dispatchers.IO，让阻塞IO在专用线程池执行：

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.reactor.mono

return mono(Dispatchers.IO) {
    coroutineScope {
        // 这里执行阻塞IO操作
    }
}.then()

或者用Reactor原生方式指定阻塞线程池：

import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers

return Mono.fromCallable {
    // 阻塞IO操作
}.subscribeOn(Schedulers.boundedElastic()).then()

关键总结 #

• WebFlux的非阻塞不是靠suspend关键字自动实现的，核心是阻塞操作必须脱离EventLoop线程。
• 优先使用WebFlux提供的响应式工具类（DataBufferUtils、ZeroCopy），避免手动操作流。
• Kotlin协程中用Dispatchers.IO，Reactor中用Schedulers.boundedElastic()处理阻塞IO。

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