油门形状的参数是什么意思？

问题描述

zio-streams提供throttleShape，

  /**
   * Delays the chunks of this stream according to the given bandwidth parameters using the token bucket
   * algorithm. Allows for burst in the processing of elements by allowing the token bucket to accumulate
   * tokens up to a `units + burst` threshold. The weight of each chunk is determined by the `costFn`
   * function.
   */
  final def throttleShape(units: Long,duration: Duration,burst: Long = 0)(
    costFn: Chunk[O] => Long
  ): ZStream[R with Clock,E,O]

我正在努力了解如何使用参数unit，duration burst和costFun。根据我对token bucket

的阅读

throttleShape(1,1.second)(_ => 1)

处理一个元素的成本为一个令牌（costFun = _ => 1），一秒钟后会补充一个令牌（unit = 1）（duration = 1.second）。但是，除了

之外，我使用各种值进行的实验似乎都不会产生任何节流作用

throttleShape(1,1.second)(_ => 2)

这使其挂起。例如，如何解释以下使用无穷持续时间的片段（来自PR）的节流

Stream(1,2,3,4)
  .throttleShape(1,Duration.Infinity)(_ => 0)
  .runCollect

Stream(1,4)
  .throttleShape(2,Duration.Infinity)(_ => 1)
  .take(2)
  .runCollect

具体来说，假设我想每分钟最多处理100个元素，那么应该如何指定throttleShape？

解决方法

问题在于您的初始流是单个Chunk[Int]，并且在throttleShape中是注释中所说的-您按块而不是按元素来限制。

单个块是根据Stream(1,2,3,4)构造的，因为它对应于

  /**
   * Creates a pure stream from a variable list of values
   */
  def apply[A](as: A*): ZStream[Any,Nothing,A] = fromIterable(as)

其中

  /**
   * Creates a stream from an iterable collection of values
   */
  def fromIterable[O](as: => Iterable[O]): ZStream[Any,O] =
    fromChunk(Chunk.fromIterable(as))

因此，如果要限制元素，则应按.chunkN(1)将块重新缩放为1个元素。您应该在节流之前这样做。

如果是

说我想每分钟最多处理100个元素

如果您不需要块的优化（以批量/块方式处理项目），则可以将块缩放到1，然后throttleShape(100,1.minute)(_ => 1)

stream.Stream.fromIterable(1 to 1000)
  .chunkN(1)
  .throttleShape(100,1.minute)(_ => 1)
  .foreachChunk(chunk => console.putStrLn(s"processed '${chunk.foldLeft("")(_ + _)}'"))

或者，如果您要分块处理并保持相同的处理速率-您可以将costFn写为_.size：

stream.Stream.fromIterable(1 to 1000)
  .chunkN(5)
  .throttleShape(100,1.minute)(_.size)
  .foreachChunk(chunk => console.putStrLn(s"processed '${chunk.foldLeft("")(_ + _)}'"))