Akka stream actor-conflation-ratelimit-actor 丢弃前几条消息(有时)
Akka stream actor-conflation-ratelimit-actor drops first few messages (sometimes)
一个简单的合并组合(下图)有时会在 staartup 上打印一条调试消息,说由于零需求它正在丢弃消息。
我希望合并阶段能提供无限的需求,所以上面的情况永远不会发生。我错过了什么?
val sourceRef = Source.actorRef[KeyedHighFreqEvent](0, OverflowStrategy.fail)
.conflateWithSeed(...into hash map...)
.throttle(8, per = 1.second, maxBurst=24, ThrottleMode.shaping)
.mapConcat(...back to individual KeyedHighFreqEvent...)
.groupedWithin(1024, 1.millisecond)
.to(Sink.actorRef(networkPublisher, Nil))
.run()
system.eventStream.subscribe(sourceRef, classOf[KeyedHighFreqEvent])
Source.actorRef
的文档对此很清楚:
The buffer can be disabled by using bufferSize
of 0 and then received messages are dropped if there is no demand
from downstream. When bufferSize
is 0 the overflowStrategy
does not matter. An async boundary is added after
this Source; as such, it is never safe to assume the downstream will always generate demand.
问题是源和合并阶段之间的异步边界。合并阶段确实提供了无限的需求,但是异步边界使得传播到源的速度变慢了。
您可以在源代码中使用缓冲区(增加 bufferSize),或者使用其他源代码,例如 Source.queue
如果合适,因为它不会引入异步边界
一个简单的合并组合(下图)有时会在 staartup 上打印一条调试消息,说由于零需求它正在丢弃消息。 我希望合并阶段能提供无限的需求,所以上面的情况永远不会发生。我错过了什么?
val sourceRef = Source.actorRef[KeyedHighFreqEvent](0, OverflowStrategy.fail)
.conflateWithSeed(...into hash map...)
.throttle(8, per = 1.second, maxBurst=24, ThrottleMode.shaping)
.mapConcat(...back to individual KeyedHighFreqEvent...)
.groupedWithin(1024, 1.millisecond)
.to(Sink.actorRef(networkPublisher, Nil))
.run()
system.eventStream.subscribe(sourceRef, classOf[KeyedHighFreqEvent])
Source.actorRef
的文档对此很清楚:
The buffer can be disabled by using
bufferSize
of 0 and then received messages are dropped if there is no demand from downstream. WhenbufferSize
is 0 theoverflowStrategy
does not matter. An async boundary is added after this Source; as such, it is never safe to assume the downstream will always generate demand.
问题是源和合并阶段之间的异步边界。合并阶段确实提供了无限的需求,但是异步边界使得传播到源的速度变慢了。
您可以在源代码中使用缓冲区(增加 bufferSize),或者使用其他源代码,例如 Source.queue
如果合适,因为它不会引入异步边界