akka-http中Multipart/FormData流如何计算ByteString大小得到总缓冲字节数

首先明确Akka HTTP中Multipart.FormData.BodyPart的dataBytes流的默认特性：
默认配置下，dataBytes输出的每个ByteString元素的大小由akka.http.parsing.max-chunk-size参数控制，默认值为8KB（8192字节），仅流的最后一个元素可能小于该值。

计算方式 #

你可以根据自己的场景选择以下两种方案：

方案1：基于默认配置快速计算 #

如果未修改过Akka HTTP的解析配置，直接用默认批次大小计算即可：
总缓冲区字节数 = buffer参数（元素个数） × 8192
示例：如果你需要总缓冲区大小为1MB，buffer参数填128即可（128 × 8KB = 1MB）

方案2：精确控制缓冲区字节大小 #

如果不想依赖默认配置，希望精确控制总缓冲区字节数，可以先对ByteString流做固定大小批次重对齐，再设置缓冲区：

import akka.util.ByteString
import akka.stream.scaladsl.Source

entity(as[Multipart.FormData]) { formData => 
  // 自定义每个批次的字节大小，比如设置为1KB
  val fixedBatchSize = 1024
  // 想要的总缓冲区字节大小，比如100KB
  val totalBufferBytes = 1024 * 100
  // 计算得到buffer的元素个数
  val bufferElementCount = totalBufferBytes / fixedBatchSize

  formData
    .parts 
    .flatMap { bodyPart => 
      bodyPart.entity.dataBytes
        // 将流重对齐为固定大小的ByteString批次
        .scan(ByteString.empty)(_ ++ _)
        .filter(_.size >= fixedBatchSize)
        .flatMapConcat { accum =>
          val (fullBatch, remaining) = accum.splitAt(fixedBatchSize)
          Source.single(fullBatch).concat(Source.single(remaining))
        }
        // 此时总缓冲区大小 = bufferElementCount * fixedBatchSize，完全可控
        .buffer(bufferElementCount, OverflowStrategy.backpressure)
    }
}

方案3：统计实际批次大小计算 #

如果你的流已经做过自定义调整，默认批次大小不适用，可以临时加统计逻辑拿到实际平均批次大小：

var totalBytes: Long = 0
var elementCount: Long = 0

bodyPart.entity.dataBytes
  // 侧路统计每个元素的大小
  .wireTap { bs =>
    totalBytes += bs.size
    elementCount += 1
  }

运行压测后即可得到平均每个元素大小 = totalBytes / elementCount，用期望的总缓冲区字节数除以该平均值，就能得到需要设置的buffer参数值。

注意事项 #

• 上述代码中的buffer是作用在所有BodyPart拼接后的整体流上，不是每个BodyPart单独分配缓冲区
• 若要避免大请求导致OOM，除了缓冲区设置外，建议同时配置Akka HTTP的akka.http.server.max-content-length参数限制请求总大小

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