Akka-http：如何向HTTP响应输出流写入大型TSV文件

我懂你遇到的问题了——想在HTTP GET请求里生成带表头的大型TSV文件，没法靠简单的JSON转TSV编组解决，还找不到直接获取Writer/OutputStream的方式，毕竟你试的HttpResponse().getWriter根本不存在😂

其实在Akka HTTP（看你路由DSL的写法应该用的是这个框架）里，它是基于异步流式模型的，不是传统Servlet那种阻塞式的处理方式，所以不能直接拿到Writer。不过我们可以通过Akka Stream的流式响应来实现需求，既避免把大型文件加载到内存，又能自定义写入表头和数据。

方法一：直接用Akka Stream Source生成TSV #

这是最推荐的方式，完全贴合Akka HTTP的流式设计，适合处理大型数据集：

import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model.{HttpResponse, HttpEntity, ContentTypes}
import akka.stream.scaladsl.Source
import akka.util.ByteString

path("export") {
  get {
    // 先获取你的业务数据
    val sampleExonRPKMs = exonRPKMService.getRPKMs(samples)
    
    // 定义TSV表头，用制表符分隔列名
    val headerLine = "SampleID\tExonID\tRPKM\n"
    // 把表头转换成Akka Stream可处理的Source
    val headerSource = Source.single(ByteString(headerLine))
    
    // 将业务数据转换成TSV行的Source，每一行都是制表符分隔的字符串
    val dataSource = Source(sampleExonRPKMs)
      .map(rpkm => s"${rpkm.sampleId}\t${rpkm.exonId}\t${rpkm.value}\n")
      .map(ByteString(_)) // 转成Akka HTTP支持的ByteString类型
    
    // 合并表头和数据的Source，先输出表头再输出数据
    val tsvSource = headerSource.concat(dataSource)
    
    // 构建流式响应，指定Content-Type为TSV格式
    complete(HttpResponse(
      entity = HttpEntity.Streamed(
        ContentTypes.`text/tab-separated-values(UTF-8)`,
        None, // 大型文件长度未知，用Chunked分块传输
        tsvSource
      )
    ))
  }
}

方法二：用Java Writer/OutputStream处理复杂写入 #

如果你需要更复杂的逻辑（比如处理特殊字符转义、用第三方库写TSV），可以通过StreamConverters获取OutputStream，再包装成Writer来用：

import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model.{HttpResponse, HttpEntity, ContentTypes}
import akka.stream.scaladsl.{Source, StreamConverters}
import java.io.{OutputStreamWriter, StandardCharsets}
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

path("export") {
  get {
    val sampleExonRPKMs = exonRPKMService.getRPKMs(samples)
    
    // 创建一个能输出到HTTP响应的OutputStream Source
    val tsvSource = StreamConverters.asOutputStream().mapMaterializedValue { outputStream =>
      // 把写入逻辑放到Future里，用单独线程处理，避免阻塞Akka的IO线程
      Future {
        val writer = new OutputStreamWriter(outputStream, StandardCharsets.UTF_8)
        try {
          // 写入表头
          writer.write("SampleID\tExonID\tRPKM\n")
          // 遍历数据写入每一行，可自定义转义逻辑
          sampleExonRPKMs.foreach { rpkm =>
            writer.write(s"${escape(rpkm.sampleId)}\t${escape(rpkm.exonId)}\t${rpkm.value}\n")
          }
        } finally {
          // 必须关闭流，否则HTTP连接会一直挂起
          writer.flush()
          writer.close()
        }
      }
    }
    
    // 构建最终响应
    complete(HttpResponse(
      entity = HttpEntity.Streamed(
        ContentTypes.`text/tab-separated-values(UTF-8)`,
        None,
        tsvSource
      )
    ))
  }
}

// 示例：自定义TSV特殊字符转义方法
def escape(s: String): String = {
  if (s.contains("\t") || s.contains("\n")) s""""${s.replace("\"", "\"\"")}"""" else s
}

关键注意点 #

1. 不要把大型数据集一次性加载到内存，用Akka Stream的Source流式处理，避免内存溢出
2. 如果用第二种方法，一定要把写入逻辑放到Future中，用非阻塞线程处理，别占用Akka核心IO线程
3. 记得指定正确的Content-Type为text/tab-separated-values(UTF-8)，确保客户端能正确识别文件类型

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