如何用fs2 Stream实现类Akka Stream的Pass Through机制？

要实现你描述的需求——将Message拆分为offset和record两个子流，分别处理后再合并回原类型，其中一个子流可透传——可以借助fs2的broadcast和zip/zipWith操作来完成，核心思路是先将源流复制为两个分支，分别处理后再按顺序一一合并。

代码实现示例 #

import fs2.{Stream, Pipe, Pure}

// 定义消息类型
case class Message(offset: Int, record: String)

// 示例：处理offset的Pipe（这里做简单的数值加10）
val offsetProcessingPipe: Pipe[Pure, Int, Int] = _.map(_ + 10)

// 示例：处理record的Pipe（这里转大写，若要透传直接用Pipe.identity即可）
val recordProcessingPipe: Pipe[Pure, String, String] = _.map(_.toUpperCase)
// val recordProcessingPipe: Pipe[Pure, String, String] = Pipe.identity // 透传版本

// 源数据流
val messageStream: Stream[Pure, Message] = Stream(
  Message(1, "hello"),
  Message(2, "world"),
  Message(3, "fs2")
)

// 核心处理流程：拆分 -> 分别处理 -> 合并
val processedStream: Stream[Pure, Message] = messageStream
  .broadcast(2) // 将源流拆分为两个完全相同的分支
  .match {
    case offsetBranch :: recordBranch :: Nil =>
      // 第一个分支处理offset
      val processedOffsets = offsetBranch.map(_.offset).through(offsetProcessingPipe)
      // 第二个分支处理record
      val processedRecords = recordBranch.map(_.record).through(recordProcessingPipe)
      // 将处理后的结果合并回Message
      processedOffsets.zipWith(processedRecords)(Message(_, _))
    case _ => Stream.empty // broadcast(2)不会返回其他分支数量，这里做兜底
  }

// 运行验证
processedStream.compile.toList.foreach(println)
// 输出：
// Message(11,HELLO)
// Message(12,WORLD)
// Message(13,FS2)

关键操作说明 #

• broadcast(n): 将源流复制为n个独立分支，每个分支都会收到源流的所有元素，保证了两个子流的元素顺序和数量完全一致，为后续合并提供基础。
• zipWith: 按顺序将两个流的元素一一配对，用给定的函数（这里是Message(_, _)）合并成新的元素，完美匹配你需要的“两个子流都发出值时再合并”的需求。
• 透传处理: 如果其中一个子流不需要处理，直接使用Pipe.identity作为对应的处理Pipe即可，它会原样传递所有元素。

封装复用 #

如果需要多次使用这个逻辑，可以将整个流程封装为一个可复用的Pipe：

def messagePassThroughPipe[F[_]](
  offsetPipe: Pipe[F, Int, Int],
  recordPipe: Pipe[F, String, String]
): Pipe[F, Message, Message] =
  _.broadcast(2) match {
    case offsetBranch :: recordBranch :: Nil =>
      val processedOffsets = offsetBranch.map(_.offset).through(offsetPipe)
      val processedRecords = recordBranch.map(_.record).through(recordPipe)
      processedOffsets.zipWith(processedRecords)(Message(_, _))
    case _ => Stream.empty
  }

// 使用封装后的Pipe
val reusedProcessedStream = messageStream.through(messagePassThroughPipe(offsetProcessingPipe, recordProcessingPipe))

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