Akka Stream 阻塞 I/O 桥接指南：StreamConverters 的 InputStream/OutputStream 适配器与陷阱规避

StreamConverters 是 Akka Stream 为 java.io.InputStream、java.io.OutputStream 以及 Java 8 Stream 提供的一组桥接算子，用于在响应式流与遗留阻塞 I/O API 之间建立单向数据通道。本文以 Akka 仓库中的官方分类文档 additional-sink-and-source-converters.md 为骨架，结合 akka-stream 模块的源码实现与测试用例，系统讲解这四个阻塞桥接算子的用法、blocking-io-dispatcher 专用线程池的配置方式，以及 asInputStream/asOutputStream 在 mapMaterializedValue 中使用时的死锁陷阱，帮助读者安全地将传统阻塞 I/O 接入 Akka Stream 管线。

一、StreamConverters：为阻塞 API 而生的桥接工具箱

Akka Stream 本身是异步、非阻塞的 Reactive Streams 实现，但现实系统中大量遗留库（如 HttpURLConnection、ZipInputStream、ObjectInputStream 等）只提供 java.io.InputStream / java.io.OutputStream 这类同步阻塞接口。为了在不重写这些库的前提下把它们接入流管线，akka-stream 在 StreamConverters.scala（Scala DSL）和 StreamConverters.scala（Java DSL）中提供了两组能力：

• 阻塞 I/O 桥接：fromInputStream、asOutputStream、fromOutputStream、asInputStream，实现流与 java.io 流对象的双向转换；
• Java 8 Stream 桥接：fromJavaStream、asJavaStream，实现流与 java.util.stream.Stream 的互操作；
• Java 8 Collector 聚合：javaCollector、javaCollectorParallelUnordered，用 Collector 做流内聚合归约。

正如官方文档所述，"Sources and sinks for integrating with java.io.InputStream and java.io.OutputStream can be found on StreamConverters"。这些算子本质上是阻塞的——它们内部通过 Await、信号量、有界队列等同步原语等待数据——因此官方文档强调："As they are blocking APIs the implementations of these operators are run on a separate dispatcher configured through the akka.stream.blocking-io-dispatcher." 也就是说，阻塞工作不会占用流默认的 akka.actor.default-dispatcher，而是被隔离到独立的专用调度器上，避免阻塞算子拖垮整个 ActorSystem 的消息处理吞吐。

二、四个 java.io 桥接算子：方向与物化值

四个算子按"桥接方向 × 流角色"两两组合，覆盖了阻塞 I/O 与 Akka Stream 之间的全部四种接入形态：

算子	DSL	签名（Scala DSL）	流角色	方向	物化值
fromInputStream	Source	(in: () => InputStream, chunkSize: Int = 8192)	Source	阻塞读 → 流	Future[IOResult]
asOutputStream	Source	(writeTimeout: FiniteDuration = 5.seconds)	Source	外部写 → 流	OutputStream
fromOutputStream	Sink	(out: () => OutputStream, autoFlush: Boolean = false)	Sink	流 → 阻塞写	Future[IOResult]
asInputStream	Sink	(readTimeout: FiniteDuration = 5.seconds)	Sink	流 → 阻塞读	InputStream

（表中签名与默认值均取自 scaladsl/StreamConverters.scala 第 46、63、79、95 行的定义；Java DSL 对应的 CompletionStage 变体见 javadsl/StreamConverters.scala。）

1. fromInputStream：把阻塞读取包装成 Source

val source: Source[ByteString, Future[IOResult]] =
  StreamConverters.fromInputStream(() => new FileInputStream("data.bin"))

• 每次 materialization 时调用工厂函数创建 InputStream，读取到的数据以 ByteString 形式逐个元素下发，每个元素大小不超过 chunkSize（默认 8192 字节，且要求 chunkSize > 0，见 InputStreamSource.scala 的 require 校验）；
• 物化值为 Future[IOResult]，携带累计读取的字节数；若下游提前取消，会以"下游未读完即失败"的语义补全异常（源码 onDownstreamFinish 分支）；
• 当 Source 被取消时，创建的 InputStream 会被自动关闭（见同一文件 closeInputStream 逻辑）。

2. asOutputStream：向外部暴露一个可写的 OutputStream

val out: OutputStream = // 由物化值获得
  Source(List(ByteString("hello"), ByteString(" world")))
    .toMat(StreamConverters.asOutputStream())(Keep.right)
    .run()

• 物化值直接是一个 OutputStream，外部代码向它 write 的数据会被送入下游流；
• 其实现 OutputStreamSourceStage.scala 用信号量（Semaphore(maxBuffer, fair = true)）计数未兑现的下游需求：write 会先 tryAcquire(writeTimeout)，拿不到许可（下游没消费）就抛出 IOException("Timed out trying to write data to stream")，close() 则通过 AsyncCallback 通知 stage 正常完成；
• 因此 writeTimeout（默认 5 秒）实际上限定了外部写方在"下游背压"时最多阻塞多久；
• 关闭该 OutputStream 会完成该 Source；反之 Source 被取消时 OutputStream 也会关闭。

3. fromOutputStream：把流写到阻塞输出流

val sink: Sink[ByteString, Future[IOResult]] =
  StreamConverters.fromOutputStream(() => new FileOutputStream("out.bin"))

• 流入的 ByteString 被写入工厂创建的 OutputStream；autoFlush = true 时每次写入字节数组后立即 flush()，默认 false；
• 流完成时关闭 OutputStream，流写入失败（OutputStream 不再可写）时 Sink 会取消上游；
• 物化值同样是 Future[IOResult]，完成时携带写入总字节数。

4. asInputStream：把流包装成可阻塞读取的 InputStream

val in: InputStream = // 由物化值获得
  source.runWith(StreamConverters.asInputStream())

• 物化值是一个 InputStream，外部代码可以像读普通文件流一样从中读取流内数据；
• 实现 InputStreamSinkStage.scala 用容量为 maxBuffer + 2 的 LinkedBlockingDeque 缓存 Data / Finished / Failed 消息，上游每推入一个元素就入队一个 Data，外部 read 阻塞等待队列，readTimeout（默认 5 秒）决定读取方最多阻塞多久；
• 流完成时 InputStream 会收到完成信号并关闭；外部关闭 InputStream 则会取消该 Sink。

关于物化值的两点细节（源码可证）：fromInputStream / fromOutputStream 的 Future<a href="https://link.gitcode.com/i/1bb90576b7b3ba3e9b7260425e760078" target="_blank">IOResult] 中，字节数只代表"已从源头读走 / 已写入目标"，不保证下游 stage 一定消费了这些字节（[scaladsl/StreamConverters.scala 的 scaladoc 明确说明）；asInputStream / asOutputStream 的内部缓冲区大小可以通过 ActorAttributes.InputBuffer 属性覆盖（两个 stage 都通过 inheritedAttributes.getInputBuffer).max 取值，默认 16）。

三、blocking-io-dispatcher：阻塞算子的专用线程池

由于这四个算子会把当前线程真正阻塞住（等待上游数据或下游消费），它们绝不能在 akka.actor.default-dispatcher 上执行。仓库的默认配置定义在 reference.conf：

akka.stream.materializer {
  # Fully qualified config path which holds the dispatcher configuration
  # or full dispatcher configuration to be used by stream operators that
  # perform blocking operations
  blocking-io-dispatcher = "akka.actor.default-blocking-io-dispatcher"
}

• 当前版本的推荐配置路径是 akka.stream.materializer.blocking-io-dispatcher，默认指向 akka.actor.default-blocking-io-dispatcher；
• 在代码里，这个默认值通过 Attributes.scala 中的 ActorAttributes.IODispatcher（值为字符串 "akka.stream.materializer.blocking-io-dispatcher"）在 materializer 创建时被读取（见 ActorMaterializer.scala 的 config.getString("blocking-io-dispatcher") 调用）；
• 每个具体算子的 scaladoc 都提示了两种覆盖方式：改全局配置，或对单个流用 ActorAttributes 覆盖，例如：

source.withAttributes(ActorAttributes.dispatcher("akka.actor.my-blocking-dispatcher"))

• 仓库同时在 akka.stream 命名空间下保留了 blocking-io-dispatcher 与 default-blocking-io-dispatcher 两个已废弃的旧配置项（reference.conf），注释明确指出它们只是为避免破坏 Akka HTTP 等引用方而保留，新代码应改用 akka.stream.materializer.blocking-io-dispatcher 或代码内的 ActorAttributes.IODispatcher 属性。

正是因为有这个独立调度器，fromInputStream 的 read 循环、asInputStream 读取方的阻塞 read()、asOutputStream 写入方的阻塞 write() 才不会与流中其他非阻塞 stage 争抢同一批线程。

四、核心陷阱：不要在 mapMaterializedValue 里消费阻塞物化值

这是官方文档用一整段 @@@ warning 强调的高危误用场景，原样复述如下（additional-sink-and-source-converters.md）：

Be aware that asInputStream and asOutputStream materialize InputStream and OutputStream respectively as blocking API implementation. They will block the thread until data will be available from upstream. Because of blocking nature these objects cannot be used in mapMaterializeValue section as it causes deadlock of the stream materialization process.

原因分析：mapMaterializedValue 回调运行在流的物化（materialization）线程上，而物化过程本身尚未完成——asInputStream / asOutputStream 对应的 stage 还没有把数据推入通道。此时在回调里直接调用 inputStream.read()（或 outputStream.write()），该线程会立刻阻塞等待一个永远不会先于物化完成而出现的数据，于是物化流程与阻塞调用互相等待，形成死锁，最终表现为超时异常。文档给出了触发该问题的代码形态：

...
.toMat(StreamConverters.asInputStream().mapMaterializedValue { inputStream =>
        inputStream.read()  // this could block forever
        ...
}).run()

正确做法是只在 mapMaterializedValue 中保存物化值引用，把实际的阻塞读/写放到另一个线程（例如 mapAsync、Future、或专门的外部工作线程）中执行：

val in: InputStream =
  source.runWith(StreamConverters.asInputStream())

// 在独立的 Future / 工作线程中消费，不要在物化回调中 read
Future {
  val data = in.readNBytes(1024)
  // ... 处理 data
}(ExecutionContext.global)

这一限制同样适用于 asJavaStream——其实现 scaladsl/StreamConverters.scala 在物化值构造的 Iterator 中通过 Await.result(queue.pull(), Inf) 无限期阻塞等待下一个元素，因此绝不能在物化回调内同步迭代该 Java Stream。

五、Java 8 Stream 与 Collector：同一工具箱的扩展能力

除 java.io 桥接外，StreamConverters 还提供 Java 8 生态互操作（这些内容同样属于官方文档所述"blocking API 实现运行于独立 dispatcher"的范畴）：

• fromJavaStream(() => ...)：把 Java 8 BaseStream（如 IntStream.rangeClosed(1, 10)）包装成 Source，按需向下游推送元素，物化值为 NotUsed；可用 Source.async 在 Java Stream 与流其余部分之间建立异步边界（scaladsl/StreamConverters.scala）；
• asJavaStream()：Sink 侧反向桥接，物化值为 java.util.stream.Stream[T]，流完成时 Java Stream 结束、关闭 Java Stream 会取消流入；同样阻塞读取线程，官方 scaladoc 明确其实现运行在 akka.stream.blocking-io-dispatcher 上；
• javaCollector：Sink 物化为 Future[R]，用 Java 8 Collector 顺序聚合归约；javaCollectorParallelUnordered(parallelism) 则在 parallelism 个 worker 间用 Balance / Merge 并行分片归约再合并（当 parallelism == 1 时退化为 javaCollector）。注意 Collector 工厂函数在多次 materialization 时会被多次调用，必须能够重复创建。

六、测试验证：仓库中的行为佐证

akka-stream 为这些算子配套了完整的 Scala / Java 测试，可作为行为契约参考：

• InputStreamSourceSpec.scala、OutputStreamSourceSpec.scala：验证 fromInputStream / asOutputStream 的读取分块、writeTimeout 超时抛 IOException、关闭语义与背压行为；
• InputStreamSinkSpec.scala、OutputStreamSinkSpec.scala：验证 asInputStream / fromOutputStream 的 readTimeout、autoFlush 与完成/取消语义；
• Java DSL 侧对应 InputStreamSinkTest.java、OutputStreamSourceTest.java、OutputStreamSinkTest.java；
• 聚合与互操作行为见 StreamConvertersSpec.scala。

七、实战检查清单

将上述内容沉淀为一份可直接对照的落地清单：

1. 选型：读阻塞源用 fromInputStream，暴露可写出口用 asOutputStream，写阻塞目标用 fromOutputStream，暴露可读入口用 asInputStream；
2. 线程隔离：确认 akka.stream.materializer.blocking-io-dispatcher 指向的调度器拥有足够的独立线程（默认 akka.actor.default-blocking-io-dispatcher），或对单个流用 ActorAttributes.dispatcher(...) 覆盖；不要依赖已废弃的 akka.stream.blocking-io-dispatcher 顶层配置项；
3. 超时与缓冲：readTimeout / writeTimeout 默认 5 秒，chunkSize 默认 8192，内部缓冲默认 16 元素，均可按业务调整；
4. 死锁红线：绝不在 mapMaterializedValue 回调中调用 InputStream.read() / OutputStream.write() 或迭代 asJavaStream 返回的 Java Stream，物化回调只负责传递引用；
5. 资源生命周期：外部关闭 InputStream / OutputStream 会反向取消/完成流，流完成/取消也会自动关闭对应的 java.io 流对象，避免手动管理造成泄漏或双重关闭。

只要守住"阻塞调用永远发生在物化完成之后、且运行在专用 dispatcher 上"这两条原则，StreamConverters 就能让 Akka Stream 与遗留 Java I/O 代码无缝共存。