Scala中流式读取HTTP响应前512字节后终止请求的方法

核心逻辑：

• 无需服务端支持Range请求头。HTTP基于TCP协议传输，只要客户端以流式方式打开响应通道，读取到目标长度字节后主动释放连接资源，底层会直接中断后续数据包接收，不会拉取完整响应体。
• 禁止使用会自动缓冲全量响应的客户端API（比如直接返回完整String、完整字节数组的方法），必须直接操作响应字节流。

方案1：JDK 11+ 原生HttpClient（无额外依赖） #

适用于不想引入第三方HTTP依赖的场景，Scala项目只要运行在JDK11及以上版本即可直接使用：

import java.net.URI
import java.net.http.{HttpClient, HttpRequest, HttpResponse}
import java.net.http.HttpResponse.BodyHandlers
import java.util.Arrays

def readFirstNBytes(url: String, n: Int = 512): Array[Byte] = {
  val client = HttpClient.newHttpClient()
  val request = HttpRequest.newBuilder()
    .uri(URI.create(url))
    .GET()
    .build()
  // 直接返回输入流的响应处理器，不会缓冲全量响应
  val response = client.send(request, BodyHandlers.ofInputStream())
  val inputStream = response.body()
  try {
    val buffer = new Array[Byte](n)
    var totalRead = 0
    // 循环读取直到凑够目标长度或者流提前结束
    while (totalRead < n) {
      val readCount = inputStream.read(buffer, totalRead, n - totalRead)
      if (readCount == -1) {
        return Arrays.copyOf(buffer, totalRead)
      }
      totalRead += readCount
    }
    buffer
  } finally {
    // 关键操作：读完直接关闭流，底层会主动终止连接，不再接收剩余数据
    inputStream.close()
  }
}

方案2：Pekko HTTP（原Akka HTTP）响应式流实现 #

适用于Pekko/Akka技术栈的项目，通过流原生的截断操作符即可实现自动终止：

import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.http.scaladsl.Http
import org.apache.pekko.http.scaladsl.model._
import org.apache.pekko.stream.scaladsl.Sink
import org.apache.pekko.util.ByteString
import scala.concurrent.Future
import scala.concurrent.Await
import scala.concurrent.duration._

implicit val system: ActorSystem = ActorSystem()
import system.dispatcher

def readFirstNBytesAsync(url: String, n: Int = 512): Future[ByteString] = {
  Http().singleRequest(HttpRequest(uri = url)).flatMap { response =>
    response.entity.dataBytes
      .take(n.toLong) // 只取前n个字节，流会自动向上游发送取消信号，停止拉取后续数据
      .reduce(_ ++ _)
      .runWith(Sink.head)
      .andThen { case _ => response.discardEntityBytes() } // 确保剩余实体被丢弃，连接正常回收
  }
}

// 同步调用示例
// val first512Bytes = Await.result(readFirstNBytesAsync("https://target-api.com/path"), 10.seconds)

方案3：Sttp客户端（全生态兼容） #

适用于使用Sttp作为统一HTTP客户端的项目，底层可适配Pekko、Fs2、ZIO、OkHttp等任意后端：

import sttp.client3._
import sttp.model.Uri

// 可替换为项目实际使用的后端实现
val backend = HttpURLConnectionBackend()
def readFirstNBytes(url: Uri, n: Int = 512): Array[Byte] = {
  val response = basicRequest
    .get(url)
    .response(asInputStreamAlways) // 直接返回输入流，不缓冲全量响应
    .send(backend)
  
  response.body match {
    case Right(is) =>
      try {
        val buffer = new Array[Byte](n)
        var totalRead = 0
        while (totalRead < n) {
          val cnt = is.read(buffer, totalRead, n - totalRead)
          if (cnt == -1) return buffer.take(totalRead)
          totalRead += cnt
        }
        buffer
      } finally {
        is.close() // 关闭流即终止后续数据接收
      }
    case Left(err) => throw new RuntimeException(s"Request failed: $err")
  }
}

避坑提示 #

• 不要使用客户端封装的asString、asByteArray等自动聚合全量响应的方法，这类方法会等待完整响应接收完毕才返回，完全无法实现提前终止的效果
• 读取完目标长度后必须在finally块中关闭流/触发流取消，否则连接会持续等待剩余响应数据，造成连接池泄漏、资源占用过高
• 无需手动添加Connection: close请求头，客户端在主动关闭响应流时会自动处理连接回收逻辑
• 如果是异步非阻塞客户端，不要用阻塞等待全量响应的方法，直接用流的截断操作符即可

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