如何在Quarkus中用Kotlin实现动态zip包的非阻塞流式响应？

核心思路 #

• 以Uni<Response>作为接口返回值，既满足Quarkus Reactive的响应式返回规范，也支持自定义响应头
• 通过协程管道桥接JDK原生ZipOutputStream与响应式流，实现无内存缓存、无磁盘写入的流式打包输出
• 采用Vert.x非阻塞WebClient下载源文件，避免阻塞IO限制并发能力

前置依赖 #

需确保项目引入以下依赖：

• io.quarkus:quarkus-resteasy-reactive-kotlin：Quarkus对Kotlin协程的支持扩展
• io.quarkus:quarkus-vertx-web-client：非阻塞HTTP客户端，用于下载源文件
• org.jetbrains.kotlinx:kotlinx-coroutines-reactive:1.5.2：适配Kotlin 1.5.30版本，实现Flow与Mutiny类型的互转
• org.jetbrains.kotlinx:kotlinx-coroutines-io:0.17.2：协程IO管道支持

完整实现代码 #

import io.smallrye.mutiny.Uni
import io.vertx.core.buffer.Buffer
import io.vertx.ext.web.client.WebClient
import io.vertx.kotlin.coroutines.await
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.io.ByteChannel
import kotlinx.coroutines.launch
import kotlinx.coroutines.reactive.asPublisher
import java.util.zip.ZipEntry
import java.util.zip.ZipOutputStream
import javax.inject.Inject
import javax.ws.rs.GET
import javax.ws.rs.Path
import javax.ws.rs.Produces
import javax.ws.rs.core.Response

data class Entry(val url: String, val name: String)

@Path("/zip")
class ZipResource {

    @Inject
    lateinit var webClient: WebClient

    @GET
    @Produces("application/zip")
    fun download(): Uni<Response> {
        val entries = listOf(
            Entry("http://link-to-a-source-file1", "file1.txt"),
            Entry("http://link-to-a-source-file2", "file2.txt"),
        )
        val contentDisposition = """attachment; filename="test.zip""""

        // 协程流式生成zip二进制流
        val zipFlow: Flow<Buffer> = flow {
            ByteChannel().use { channel ->
                // 启动子协程写入zip内容
                launch {
                    ZipOutputStream(channel.toOutputStream()).use { zipOut ->
                        entries.forEach { entry ->
                            zipOut.putNextEntry(ZipEntry(entry.name))
                            // 非阻塞下载源文件，边下载边写入zip
                            val resp = webClient.getAbs(entry.url).send().await()
                            resp.bodyAsFlow().collect { byteBuf ->
                                zipOut.write(byteBuf.bytes)
                                byteBuf.release()
                            }
                            zipOut.closeEntry()
                        }
                    }
                }
                // 读取管道内容发射到响应流
                while (!channel.isClosedForRead) {
                    val buf = channel.readRemaining(8 * 1024)
                    emit(Buffer.buffer(buf.readBytes()))
                }
            }
        }

        // 包装为标准Response对象，设置响应头
        return Uni.createFrom().item(
            Response.ok(zipFlow.asPublisher())
                .header("Content-Disposition", contentDisposition)
                .build()
        )
    }
}

实现说明 #

• 全程非阻塞：文件下载、zip打包、响应输出均运行在协程调度逻辑中，不会阻塞Quarkus事件循环线程，并发能力不受工作线程数量限制
• 无额外资源开销：所有内容边下载、边打包、边输出，不会将完整源文件或zip包加载到内存，也不会写入临时磁盘文件
• 无强框架绑定：核心逻辑基于Kotlin协程标准API实现，仅少量注入逻辑依赖Quarkus基础特性，迁移成本低

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