Sink fold for akka stream Source.actorRef buffer and Overflo

Question

这是来自 akka 文档的代码片段

val sinkUnderTest = Flow[Int].map(_.toString).toMat(Sink.fold("")(_ + _))(Keep.right)
val (ref, future) = Source.actorRef(3, OverflowStrategy.fail)
.toMat(sinkUnderTest)(Keep.both).run()
ref ! 1
ref ! 2
ref ! 3
ref ! akka.actor.Status.Success("done")
val result = Await.result(future, 3.seconds)
assert(result == "123")

这是一个工作代码片段，但是，如果我使用 ref 告诉另一条消息，例如ref ! 4，我得到了一个异常，例如akka.stream.BufferOverflowException: Buffer overflow (max capacity was: 3)

我想缓冲区大小 3 应该足够了。原因是折叠操作是(acc，ele)=> acc，所以需要累加器和元素来返回新的值累加器。

所以我更改了代码，让另一个演员告诉等待 3 秒。它又开始工作了。

val sinkUnderTest = Flow[Int].map(_.toString).toMat(Sink.fold("")(_ + _))(Keep.right)
private val (ref, future): (ActorRef, Future[String]) = Source.actorRef(3, OverflowStrategy.backpressure).toMat(sinkUnderTest)(Keep.both).run()
ref ! 1
ref ! 2
ref ! 3
Thread.sleep(3000)
ref ! 4
ref ! akka.actor.Status.Success("done")
val result = Await.result(future, 10.seconds)
println(result)

但是，我的问题是，有没有办法告诉 Akka 流放慢速度或等待水槽可用。我也在使用OverflowStrategy.backpressure，但它说Backpressure overflowStrategy not supported.

还有其他选择吗？

Answer 1

您应该将Source.queue视为以背压感知方式从外部将元素注入流中的一种方法。

Source.queue将具体化为可以向其提供元素的队列对象，但是当您提供元素时，您将获得一个在流准备好接受消息时完成的Future。

示例如下：

val sinkUnderTest = Flow[Int].map(_.toString).toMat(Sink.fold("")(_ + _))(Keep.right)
val (queue, future): (SourceQueueWithComplete[Int], Future[String]) =
Source.queue(3, OverflowStrategy.backpressure).toMat(sinkUnderTest)(Keep.both).run()
Future.sequence(Seq(
queue.offer(1),
queue.offer(2),
queue.offer(3),
queue.offer(4)
))
queue.complete()
val result = Await.result(future, 10.seconds)
println(result)

文档中的详细信息。