如何在ZIO Streams中实现类似Akka Stream mergeSorted的流合并排序？

在ZIO Streams中，直接提供了与Akka Stream mergeSorted 功能完全对齐的同名算子 mergeSorted，专门用于合并多个已预排序的流，并输出全局保持时序/排序规则的结果，非常适合处理历史时序数据的场景。

基础用法示例 #

假设你有两个已按升序排列的数字流，合并后输出全局有序的流：

import zio.stream._
import zio._

val stream1 = ZStream(1, 3, 5)
val stream2 = ZStream(2, 4, 6)

val mergedSortedStream = stream1.mergeSorted(stream2)

// 运行后输出：1, 2, 3, 4, 5, 6

自定义排序规则 #

如果需要基于自定义的排序逻辑合并流（比如降序或自定义对象排序），可以传入Ordering实例：

case class Event(timestamp: Long, data: String)

// 按时间戳降序排序
val eventOrdering: Ordering[Event] = Ordering.by(_.timestamp).reverse

val streamA = ZStream(Event(1000, "A1"), Event(3000, "A3"))
val streamB = ZStream(Event(2000, "B2"), Event(4000, "B4"))

val mergedEvents = streamA.mergeSorted(streamB)(eventOrdering)
// 输出顺序：Event(4000, "B4"), Event(3000, "A3"), Event(2000, "B2"), Event(1000, "A1")

核心前提 #

和Akka的mergeSorted一样，ZIO的mergeSorted要求所有输入流本身必须是按照指定排序规则预先有序的，否则无法保证输出的全局有序性。

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