2

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从我对令牌桶的阅读中

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, 2, 3, 4)
  .throttleShape(2, Duration.Infinity)(_ => 1)
  .take(2)
  .runCollect

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

4

1 回答 1

3

问题是你的初始流是一个单一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, Nothing, O] =
    fromChunk(Chunk.fromIterable(as))

因此，如果您想按元素节流，您应该将块重新缩放为 1 个元素.chunkN(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("")(_ + _)}'"))
于 2020-08-27T22:36:41.927 回答