throttleShape 的参数是什么意思？

Question

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)

这使它挂起。例如，如何解释以下使用无限持续时间的 sn-ps（来自 PR）中的节流

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

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

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

Answer 1

问题在于，您的初始流是单个 Chunk[Int] 和 throttleShape，正如它在 cmets 中所说的那样 - 您按块进行节流，而不是按元素。

单个块是从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, Nothing, 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("")(_ + _)}'"))