Queue

Queue

Queue 是一个基于 ZIO 的轻量级的,驻扎在内存的队列,具有可组合性,并且支持透明的背压。它是完全异步(无锁或无阻塞)的,纯函数且类型安全的。 Queue[A] 包含的值的类型为 A,并且具有两个基本操作:offer,将 A 放入队列中,和 take, 删除并返回队列中最旧的值。 新建一个队列 一个 Queue 可以是有界的(容量有限)或无界的。 当队列已满时,有几种策略可用于处理新值: 对默认的有界(bounded)队列是反压:填满后,任何提供数据的纤程都将被挂起,直到队列能够添加新内容为止; 下降(dropping)队列当队列满时会丢弃新的数据; 当滑动(sliding)队列满时会丢弃最陈旧的数据。 建立一个支持背压的有界队列: 建立一个下降队列: 建立一个滑动队列: 建立一个无界队列: 向队列添加数据 向队列添加值的最简单方法是 offer: 使用背压队列时,如果队列已满,offer 可能会暂时挂起:您可以使用 fork 让其在其它纤程中等待。 还可以使用 offerAll 一次添加多个值: 从队列中消费数据 take 操作从队列中删除最旧的数据并返回它。如果队列为空,它将时挂起,直到新的数据添加到队列后才继续。与 offer 一样,您可以使用 fork 让其在其他纤程中等待新值。 您可以通过 poll 来消费最旧的数据。如果队列为空,你会得到 None,否则数据将被包装在 Some 返回。 可以使用 takeUpTo 一次消费多个数据。如果队列中没有足够的数据要退回,则它将返回所有数据,而无需等待更多 offer 。 同样,您可以使用 takeAll 一次获得所有数据。它也无需等待就返回(如果队列为空,则为空列表)。 关闭队列 可以通过 shutdown 来中断(interrupt exception)所有处于 offer* 或 take* 而挂起的纤程。它还将清空队列,并让所有未来的 offer* 和 take* 调用立刻结束(interrupt exception)。 您可以在队列关闭时使用 awaitShutdown 执行 effect。它将等待直到队列被关闭。如果队列已经关闭,它将立即返回。 转换队列Transforming queues 实际上,Queue[A] 是 ZQueue[Any, Any, Nothing, Nothing, A, A] 类型的别名。完整版本的签名为: 它的含义是: 这个队列接收 A 类型的值。需要 RA 类型的环境支持入队操作,并且可能会因EA 类型的错误而失败; 队列将产生 B 类型输出值。需要 RB 型环境支持出队操作,并可能因 EB 型错误而失败。 请注意,基本的 Queue[A] 不会失败也不需要任何环境来支持进行任何操作。 基于输入和输出的不同类型参数,可以实现各种不同的队列组合: ZQueue#map 可以将队列的输出映射(到不同类型): ZQueue#mapM 我们还可以使用一个效果(effectful)函数来映射输出。例如,我们可以为每个元素加上出列的时间戳: ZQueue#contramapM 与 mapM 相似,我们也可以在元素入列时将效果(effectful)函数应用于元素。此例将使用其入队时间戳注释元素: 此例中的队列与上一个例子中的队列具有相同的类型((Long, String)),但是当元素排队时,时间戳附加到元素上。这反映在排队入队所需的环境类型中。 一个完整的示例:我们可以将此队列(contramapM)与 mapM 结合起来,以计算元素保留在队列中的时长: ZQueue#bothWith 我们也可以将两个队列组合成一个队列,该(组合)队列广播 offer (到两个队列中)并(同时)从两个队列中接收(take)消息: 附加资料 ZIO Queue Talk by John De Goes @ ScalaWave 2018 ZIO Queue Talk by Wiem Zine El Abidine @ PSUG 2018 Elevator Control System using ZIO Scalaz 8 IO vs Akka (typed) actors vs Monix

BACK TO TOP