如何编组List[HttpResponse]？批量获取图片API实现遇阻求助

错误原因 #

Akka HTTP的complete方法要求传入的结果能被编组为单个HttpResponse，但你返回的List[HttpResponse]没有默认的Marshaller——HTTP协议本身不支持一次请求返回多个独立响应，所以直接返回列表会触发编组失败。

解决方案：将多图片打包为合法的HTTP响应 #

方案1：返回多部分（Multipart）响应 #

把每个图片作为Multipart响应的一个部分，客户端可以逐个解析这些部分。

修改代码如下：

import akka.http.scaladsl.model._
import akka.http.scaladsl.marshalling.Marshal
import akka.stream.scaladsl.Source

def getImagesFromIds(idsList: List[String]): Future[Multipart.FormData] = {
  // 将每个图片请求转为Multipart.BodyPart
  val bodyPartsFuture: Future[List[Multipart.FormData.BodyPart]] = Future.traverse(idsList) { id =>
    getImageById(id).map { response =>
      // 从原响应中提取图片内容和类型
      val entity = response.entity
      Multipart.FormData.BodyPart(
        name = s"image-$id",
        entity = entity,
        headers = List(ContentDisposition.formData(s"image-$id", Map("filename" -> s"$id.png")))
      )
    }
  }

  // 将BodyPart列表转为Multipart.FormData
  bodyPartsFuture.map { parts =>
    Multipart.FormData(Source(parts))
  }
}

// 更新路由
def getImagesByIdsListRoute: Route = get {
  path("by-ids-list") {
    entity(as[List[String]]) { upcs =>
      complete(getImagesFromIds(upcs))
    }
  }
}

方案2：返回ZIP压缩包（推荐大图片场景） #

把所有图片打包成ZIP文件返回，客户端下载后解压即可。

需要引入ZIP处理依赖（比如commons-compress），示例代码：

import org.apache.commons.compress.archivers.zip.ZipArchiveEntry
import org.apache.commons.compress.archivers.zip.ZipArchiveOutputStream
import akka.http.scaladsl.model.HttpEntity.Strict
import akka.util.ByteString
import java.io.ByteArrayOutputStream

def getImagesFromIds(idsList: List[String]): Future[HttpResponse] = {
  // 先获取所有图片的严格实体（小图片适用；大图片建议用流式处理）
  val imageEntities: Future[List[(String, Strict)]] = Future.traverse(idsList) { id =>
    getImageById(id).flatMap(_.entity.toStrict(5.seconds)).map(entity => (id, entity))
  }

  imageEntities.map { images =>
    // 构建ZIP字节流
    val byteArrayOutputStream = new ByteArrayOutputStream()
    val zipOutputStream = new ZipArchiveOutputStream(byteArrayOutputStream)
    
    images.foreach { case (id, entity) =>
      val entry = new ZipArchiveEntry(s"$id.png")
      zipOutputStream.putArchiveEntry(entry)
      val bytes = entity.dataBytes.runFold(ByteString.empty)(_ ++ _).get
      zipOutputStream.write(bytes.toArray)
      zipOutputStream.closeArchiveEntry()
    }
    zipOutputStream.close()

    // 返回ZIP响应
    HttpResponse(
      entity = HttpEntity(ContentTypes.`application/zip`, ByteString.fromArray(byteArrayOutputStream.toByteArray))
    )
  }
}

// 路由无需修改，直接返回Future[HttpResponse]即可
def getImagesByIdsListRoute: Route = get {
  path("by-ids-list") {
    entity(as[List[String]]) { upcs =>
      complete(getImagesFromIds(upcs))
    }
  }
}

方案3：返回JSON格式（小图片场景） #

将图片转为Base64编码，放在JSON数组中返回（仅适合小图片，避免响应过大）：

import spray.json._
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._

// 定义JSON格式
case class ImageData(id: String, content: String, contentType: String)
object ImageJsonProtocol extends DefaultJsonProtocol {
  implicit val imageDataFormat: RootJsonFormat[ImageData] = jsonFormat3(ImageData)
}
import ImageJsonProtocol._

def getImagesFromIds(idsList: List[String]): Future[List[ImageData]] = {
  Future.traverse(idsList) { id =>
    getImageById(id).flatMap { response =>
      response.entity.toStrict(5.seconds).map { strictEntity =>
        val base64Content = strictEntity.dataBytes.runFold(ByteString.empty)(_ ++ _).get.encodeBase64
        ImageData(id, base64Content, response.entity.contentType.toString())
      }
    }
  }
}

// 路由直接返回JSON列表
def getImagesByIdsListRoute: Route = get {
  path("by-ids-list") {
    entity(as[List[String]]) { upcs =>
      complete(getImagesFromIds(upcs))
    }
  }
}

批量请求逻辑优化 #

原代码用Future.sequence会同时发起所有请求，可能压垮目标服务器或触发限流。建议控制并发数：

方式1：用Akka Stream控制并发 #

import akka.stream.scaladsl.{Sink, Source}

def getImagesFromIds(idsList: List[String], maxConcurrency: Int = 5): Future[List[HttpResponse]] = {
  Source(idsList)
    .mapAsync(maxConcurrency)(id => getImageById(id))
    .runWith(Sink.seq)
    .map(_.toList)
}

方式2：手动分批次处理 #

如果不用Akka Stream，可手动分批次发起请求：

def batchProcess[A, B](list: List[A], batchSize: Int)(f: A => Future[B]): Future[List[B]] = {
  list.grouped(batchSize).foldLeft(Future.successful(List.empty[B])) { (acc, batch) =>
    acc.flatMap { results =>
      Future.traverse(batch)(f).map(results ++ _)
    }
  }
}

// 使用示例：每次批量5个请求
def getImagesFromIds(idsList: List[String]): Future[List[HttpResponse]] = {
  batchProcess(idsList, 5)(getImageById)
}

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