当发布商处理每个订户的压力时，Akka流是否会下沉？

Question

我将一个流作为发布者运行，并将扇出设置为true。

现在，如果我从这个发布者开始启动多个流，并且它们具有不同的处理速率，那么akka会分别处理每个用户的背压吗？

我编了一个测试：

  val publisher: Publisher[Int] = Source.fromIterator{ () => Stream.from(1).iterator}.throttle(1, 1.second).runWith(Sink.asPublisher(true))
  // start the publisher
  Source.fromPublisher(publisher).runWith(Sink.ignore)

  Thread.sleep(2000)
  Source.fromPublisher(publisher).throttle(1, 2.seconds).runForeach(x => Logger.info(s"AAAAA: $x"))
  Source.fromPublisher(publisher).throttle(1, 4.seconds).runForeach(x => Logger.info(s"BBBBB: $x"))
  Source.fromPublisher(publisher).throttle(1, 6.seconds).runForeach(x => Logger.info(s"CCCCC: $x"))

原来答案是真的：

[info] application - CCCCC: 3
[info] application - BBBBB: 3
[info] application - AAAAA: 3
[info] application - AAAAA: 4
[info] application - AAAAA: 5
[info] application - BBBBB: 4
[info] application - AAAAA: 6
[info] application - CCCCC: 4
[info] application - AAAAA: 7
[info] application - BBBBB: 5
[info] application - AAAAA: 8
[info] application - CCCCC: 5
[info] application - BBBBB: 6
[info] application - AAAAA: 9
[info] application - AAAAA: 10
[info] application - BBBBB: 7
[info] application - AAAAA: 11
[info] application - AAAAA: 12

看起来发布者为每个订阅者提供一个缓冲区并分别处理它们的背压。我对吗？

Answer 1

我不认为是这种情况。虽然我无法解释它，但是具有假定为高速率的流的应用的修改版本不能产生期望的快速速率。

import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import akka.stream.scaladsl.{Sink, Source}
import org.reactivestreams.Publisher

import concurrent.duration._

object StreamsPublisher extends App {

  implicit private val actorSystem = ActorSystem()
  private val value = ActorMaterializerSettings(actorSystem).withSupervisionStrategy{ex =>
    ex.printStackTrace()
    Supervision.Resume
  }
  private val materializer = ActorMaterializer(value)
  implicit private val mater = materializer


  val publisher: Publisher[Int] = Source
    .fromIterator { () =>
      println("new iterator created")
      Stream.from(1).iterator
    }
    .runWith(Sink.asPublisher(true))

  Source.fromPublisher(publisher).runWith(Sink.ignore)

  Thread.sleep(2000)
  Source
    .fromPublisher(publisher)
    .throttle(1, 2.seconds)
    .runForeach(x => println(s"AAAAA: $x"))
  Thread.sleep(3000)
  Source
    .fromPublisher(publisher)
    .throttle(1, 4.seconds)
    .runForeach(x => println (s"BBBBB: $x"))
  Thread.sleep(5000)
  Source
    .fromPublisher(publisher)
    .throttle(1, 6.seconds)
    .runForeach(x => println(s"CCCCC: $x"))
  Source
    .fromPublisher(publisher)
    .runForeach(x => println(s"DDDDDD: $x"))

}

“快速”流“DDDDD”运行速度不快。

输出是

new iterator created
AAAAA: 440617
AAAAA: 440618
BBBBB: 440633
AAAAA: 440619
AAAAA: 440620
BBBBB: 440634
CCCCC: 440633
DDDDDD: 440633
DDDDDD: 440634
DDDDDD: 440635
DDDDDD: 440636
DDDDDD: 440637
DDDDDD: 440638
DDDDDD: 440639
DDDDDD: 440640
DDDDDD: 440641
DDDDDD: 440642
DDDDDD: 440643
DDDDDD: 440644
DDDDDD: 440645
DDDDDD: 440646
DDDDDD: 440647
DDDDDD: 440648
AAAAA: 440621
AAAAA: 440622
BBBBB: 440635
AAAAA: 440623
DDDDDD: 440649
DDDDDD: 440650
DDDDDD: 440651
DDDDDD: 440652
DDDDDD: 440653
DDDDDD: 440654
DDDDDD: 440655
DDDDDD: 440656
AAAAA: 440624
CCCCC: 440634
BBBBB: 440636
AAAAA: 440625
AAAAA: 440626
BBBBB: 440637
AAAAA: 440627
CCCCC: 440635
AAAAA: 440628
BBBBB: 440638
AAAAA: 440629
AAAAA: 440630
CCCCC: 440636
BBBBB: 440639
AAAAA: 440631
DDDDDD: 440657
DDDDDD: 440658
DDDDDD: 440659
DDDDDD: 440660
DDDDDD: 440661
DDDDDD: 440662
DDDDDD: 440663
DDDDDD: 440664
AAAAA: 440632
BBBBB: 440640
AAAAA: 440633
CCCCC: 440637

请注意，“忽略”流很快并在加入限制流之前迭代到440K。

看起来存在小尺寸的缓冲器，这使得油门依赖。