Akka Stream性能远低于ArrayBlockingQueue双线程方案，求排查

咱们先拆解一下你的问题：你的Akka Stream实现确实存在几个可优化的细节，导致它和原生双线程队列的性能差距拉得这么大。先说说原生版本为啥快——它的逻辑极简：生产者直接把复用的char数组丢进队列，消费者线程拿了就写BufferedWriter，几乎没有额外开销，连GC压力都极小（因为数组复用）。而Akka Stream为了提供背压、容错、流组合这些强大特性，本身会有一些调度和消息传递的开销，但只要优化到位，性能应该能接近原生实现。

下面是具体的优化点和调整建议：

1. 避免不必要的async边界 #

你在buffer之后加了.async，这会强制把Source和Sink放到不同的Actor执行上下文里，增加了Actor间消息传递的开销。如果你的场景不需要严格隔离线程（比如不需要把Source和Sink分别绑定到不同调度器），直接去掉.async就能减少不少线程切换的成本。如果确实需要隔离，也要确保给IO密集型的阶段指定专门的IO调度器（比如akka.stream.blocking-io-dispatcher），而不是用默认的调度器。

2. 复用ByteString实例，减少GC压力 #

你的Source.repeat(ByteString("0123456789"))会每次生成一个新的ByteString对象，频繁创建小对象会触发频繁的GC，拖慢性能。而原生版本复用了同一个char数组，几乎没有对象分配。改成复用同一个ByteString实例：

val data = ByteString("0123456789")
Source.cycle(() => Iterator.single(data))

这样整个流只会用同一个ByteString对象，GC压力会大幅降低。

3. 调整Buffer配置，匹配IO批次 #

你给Source和Sink都设置了8192的buffer，但这个数值不一定是最优的。对于磁盘IO来说，更大的buffer（比如32768或64KB）能减少系统调用的次数。另外，FileIO.toPath默认的写入缓冲区可能比较小，你可以通过调整OpenOptions或者给Sink配置更大的输入buffer来优化：

FileIO.toPath(Paths.get("test.txt"))
  .withAttributes(Attributes.inputBuffer(32768, 32768))

4. 给IO阶段指定合适的调度器 #

Akka Stream的默认调度器是为CPU密集型任务设计的，对于磁盘IO这种阻塞型操作，应该用专门的blocking-io-dispatcher，避免阻塞默认调度器的线程池。可以给Source和Sink都加上这个属性：

.async(Attributes.dispatcher("akka.stream.blocking-io-dispatcher"))
// 或者给Sink单独设置
FileIO.toPath(...)
  .withAttributes(Attributes.dispatcher("akka.stream.blocking-io-dispatcher"))

5. 优化自定义BufferedWriter Sink #

如果你用自定义Sink，记得给BufferedWriter设置足够大的缓冲区（默认只有8KB），比如：

val writer = Files.newBufferedWriter(
  Paths.get("test.txt"),
  StandardOpenOption.CREATE,
  StandardOpenOption.WRITE,
  StandardOpenOption.TRUNCATE_EXISTING
)
// 设置64KB的缓冲区
val bufferedWriter = new BufferedWriter(writer, 65536)

同时，尽量用批量写入的方式，而不是每次写10字节，比如攒够一定数量的ByteString再一次性写入，减少IO调用次数。

优化后的示例代码 #

把这些优化点整合起来，你可以试试这个版本：

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._
import akka.util.ByteString
import java.nio.file.{Paths, StandardOpenOption}

object OptimizedAkkaStreamTest extends App {
  implicit val system: ActorSystem = ActorSystem("TestActorSystem")
  implicit val materializer: Materializer = Materializer(system)

  // 复用同一个ByteString实例
  val data = ByteString("0123456789")
  val fileOptions = Set(
    StandardOpenOption.CREATE,
    StandardOpenOption.WRITE,
    StandardOpenOption.TRUNCATE_EXISTING
  )

  Source.cycle(() => Iterator.single(data))
    // 更大的buffer，减少背压交互次数
    .buffer(32768, OverflowStrategy.backpressure)
    // 用IO调度器处理Source的生成（如果需要）
    .async(Attributes.dispatcher("akka.stream.blocking-io-dispatcher"))
    .runWith(
      FileIO.toPath(Paths.get("test.txt"), fileOptions)
        .withAttributes(
          Attributes.inputBuffer(32768, 32768)
            .and(Attributes.dispatcher("akka.stream.blocking-io-dispatcher"))
        )
    )
}

最后说下测试公平性 #

还要注意测试的公平性：比如每次测试前清空磁盘缓存（避免缓存影响结果），确保两个测试的JVM参数一致（比如堆大小、GC参数），并且写入足够大的文件（让缓存饱和，体现真实的磁盘IO性能）。

按照这些优化点调整后，Akka Stream的性能应该能大幅提升，接近原生双线程版本的水平。毕竟Akka Stream的设计目标之一就是高性能，只是需要根据场景调整配置和实现细节~

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