基于Scala实现大文件按固定条数拆分并批量调用REST API

嘿，这个场景我熟！用Scala处理大文件批量调用API的话，核心就是把文件拆成100条一批，然后逐个批次发请求就行。下面是完整的实现方案，包含代码和关键细节：

1. 依赖准备 #

首先得引入必要的库，这里用Akka HTTP来处理REST请求（Scala生态里常用的HTTP客户端），在你的build.sbt里添加依赖：

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-http" % "10.2.10",
  "com.typesafe.akka" %% "akka-stream" % "2.6.20"
)

2. 核心实现代码 #

下面是完整的可运行代码，包含文件读取、分批、API调用的逻辑：

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.scaladsl._
import akka.util.ByteString
import java.nio.file.Paths
import scala.concurrent.{ExecutionContext, Future}

object BatchApiProcessor {
  def main(args: Array[String]): Unit = {
    // 初始化Akka系统和执行上下文（用于HTTP请求和异步处理）
    implicit val system: ActorSystem = ActorSystem("BatchApiProcessor")
    implicit val ec: ExecutionContext = system.dispatcher

    // 配置参数 - 替换成你的实际路径和API地址
    val filePath = "/path/to/your/large-file.txt"
    val apiEndpoint = "https://your-api-domain.com/process-records"
    val batchSize = 100

    // 流式读取文件（适合超大文件，避免内存溢出）
    FileIO.fromPath(Paths.get(filePath))
      // 按换行符拆分每行记录
      .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 4096, allowTruncation = true))
      .map(_.utf8String.trim) // 转成字符串并去除首尾空格
      .filter(_.nonEmpty) // 跳过空行
      .grouped(batchSize) // 按100条拆分批次
      // 控制并发请求数（比如同时最多发5个请求，避免压垮API）
      .mapAsync(parallelism = 5) { batch =>
        sendBatchToApi(batch.toList, apiEndpoint)
      }
      .runForeach { response =>
        response.status match {
          case StatusCodes.OK => println(s"批次处理成功！")
          case other => println(s"批次处理失败，状态码：$other")
        }
        // 释放响应资源
        response.discardEntityBytes()
      }
      .onComplete { _ =>
        // 所有批次处理完成后关闭Akka系统
        system.terminate()
      }
  }

  /**
   * 把单个批次的记录发送到API
   * @param batch 100条记录的列表
   * @param apiUrl API的端点地址
   * @return API响应的Future
   */
  private def sendBatchToApi(batch: List[String], apiUrl: String)(implicit system: ActorSystem, ec: ExecutionContext): Future[HttpResponse] = {
    // 把记录转换成API需要的JSON格式（这里假设API接受{"records": ["记录1", "记录2"...]}的结构）
    val jsonBody = s"""{"records": [${batch.map(r => s""""$r"""").mkString(",")}]}"""
    val requestEntity = HttpEntity(ContentTypes.`application/json`, ByteString(jsonBody))

    // 构建POST请求
    val request = HttpRequest(
      method = HttpMethods.POST,
      uri = apiUrl,
      entity = requestEntity
    )

    // 发送请求并返回响应
    Http().singleRequest(request)
  }
}

3. 关键细节说明 #

• 流式处理优化：用Akka Stream的FileIO流式读取文件，不会把整个10000行文件一次性加载到内存，就算文件再大也不会OOM，适合生产环境。
• 并发控制：mapAsync(parallelism = 5)控制同时发送的API请求数，避免一次性发太多请求触发API的限流机制，这个数值可以根据API的并发限制调整。
• 错误处理：示例里只做了简单的状态码判断，实际生产中可以添加重试逻辑（比如用Akka的retry模式），把失败的批次记录到日志或者死信队列，后续重新处理。
• 请求格式适配：代码里假设API接受JSON数组格式，你需要根据实际API的文档调整请求体的结构，比如如果API接受表单数据，就把ContentTypes换成application/x-www-form-urlencoded，并调整请求体格式。
• 空行过滤：filter(_.nonEmpty)跳过文件里的空行，避免无效请求。

内容的提问来源于stack exchange，提问作者A.Mishra