在这个 akka 流示例的运行结果中，为什么优先级作业在正常作业之后出现？

Question

folling示例来自akka流引用文档。

import akka.actor.ActorSystem
import akka.stream._
import akka.stream.scaladsl._
/**
  * Created by lc on 2016/1/2.
  */
// A shape represents the input and output ports of a reusable
// processing module
case class PriorityWorkerPoolShape[In, Out](
                                             jobsIn: Inlet[In],
                                             priorityJobsIn: Inlet[In],
                                             resultsOut: Outlet[Out]) extends Shape {
  // It is important to provide the list of all input and output
  // ports with a stable order. Duplicates are not allowed.
  override val inlets: scala.collection.immutable.Seq[Inlet[_]] =
    jobsIn :: priorityJobsIn :: Nil
  override val outlets: scala.collection.immutable.Seq[Outlet[_]] =
    resultsOut :: Nil
  // A Shape must be able to create a copy of itself. Basically
  // it means a new instance with copies of the ports
  override def deepCopy() = PriorityWorkerPoolShape(
    jobsIn.carbonCopy(),
    priorityJobsIn.carbonCopy(),
    resultsOut.carbonCopy())
  // A Shape must also be able to create itself from existing ports
  override def copyFromPorts(
                              inlets: scala.collection.immutable.Seq[Inlet[_]],
                              outlets: scala.collection.immutable.Seq[Outlet[_]]) = {
    assert(inlets.size == this.inlets.size)
    assert(outlets.size == this.outlets.size)
    // This is why order matters when overriding inlets and outlets.
    PriorityWorkerPoolShape[In, Out](inlets(0).as[In], inlets(1).as[In], outlets(0).as[Out])
  }
}
import akka.stream.FanInShape.{Init, Name}
class PriorityWorkerPoolShape2[In, Out](_init: Init[Out] = Name("PriorityWorkerPool"))
  extends FanInShape[Out](_init) {
  protected override def construct(i: Init[Out]) = new PriorityWorkerPoolShape2(i)
  val jobsIn = newInlet[In]("jobsIn")
  val priorityJobsIn = newInlet[In]("priorityJobsIn")
  // Outlet[Out] with name "out" is automatically created
}
object PriorityWorkerPool {
  def apply[In, Out](
                      worker: Flow[In, Out, Any],
                      workerCount: Int): Graph[PriorityWorkerPoolShape[In, Out], Unit] = {
    FlowGraph.create() { implicit b ⇒
      import FlowGraph.Implicits._
      val priorityMerge = b.add(MergePreferred[In](1))
      val balance = b.add(Balance[In](workerCount))
      val resultsMerge = b.add(Merge[Out](workerCount))
      // After merging priority and ordinary jobs, we feed them to the balancer
      priorityMerge ~> balance
      // Wire up each of the outputs of the balancer to a worker flow
      // then merge them back
      for (i <- 0 until workerCount)
        balance.out(i) ~> worker ~> resultsMerge.in(i)
      // We now expose the input ports of the priorityMerge and the output
      // of the resultsMerge as our PriorityWorkerPool ports
      // -- all neatly wrapped in our domain specific Shape
      PriorityWorkerPoolShape(
        jobsIn = priorityMerge.in(0),
        priorityJobsIn = priorityMerge.preferred,
        resultsOut = resultsMerge.out)
    }
  }
}

object ReusableGraph extends App {
  implicit val system = ActorSystem("UsingGraph")
  implicit val materializer = ActorMaterializer()
  val worker1 = Flow[String].map("step 1 " + _)
  val worker2 = Flow[String].map("step 2 " + _)
  RunnableGraph.fromGraph(FlowGraph.create() { implicit b =>
    import FlowGraph.Implicits._
    val priorityPool1 = b.add(PriorityWorkerPool(worker1, 4))
    val priorityPool2 = b.add(PriorityWorkerPool(worker2, 2))
    Source(1 to 10).map("job: " + _) ~> priorityPool1.jobsIn
    Source(1 to 10).map("priority job: " + _) ~> priorityPool1.priorityJobsIn
    priorityPool1.resultsOut ~> priorityPool2.jobsIn
    Source(1 to 10).map("one-step, priority " + _) ~> priorityPool2.priorityJobsIn
    priorityPool2.resultsOut ~> Sink.foreach(println)
    ClosedShape
  }).run()
}

build.sbt

name := "AkkaStream"
version := "1.0"
scalaVersion := "2.11.7"
libraryDependencies ++=Seq(
  "com.typesafe.akka" % "akka-actor_2.11" % "2.4.1",
  "com.typesafe.akka" % "akka-testkit_2.11" % "2.4.1",
  "com.typesafe.akka" % "akka-stream-experimental_2.11" % "2.0-M2"
)

我运行代码，得到如下结果。

step 2 one-step, priority 1
step 2 one-step, priority 3
step 2 one-step, priority 2
step 2 one-step, priority 5
step 2 one-step, priority 4
step 2 one-step, priority 6
step 2 one-step, priority 7
step 2 one-step, priority 8
step 2 one-step, priority 10
step 2 one-step, priority 9
step 2 step 1 job: 2
step 2 step 1 job: 1
step 2 step 1 job: 4
step 2 step 1 job: 6
step 2 step 1 job: 8
step 2 step 1 job: 10
step 2 step 1 priority job: 2
step 2 step 1 priority job: 4
step 2 step 1 priority job: 6
step 2 step 1 priority job: 8
step 2 step 1 priority job: 10
step 2 step 1 job: 3
step 2 step 1 job: 5
step 2 step 1 job: 7
step 2 step 1 job: 9
step 2 step 1 priority job: 1
step 2 step 1 priority job: 3
step 2 step 1 priority job: 5
step 2 step 1 priority job: 7
step 2 step 1 priority job: 9

我有两个问题：
1.第二步第一步，是的。但是"第2步第1项工作"应该在"第2步骤第1项优先工作"之后，为什么它在"第1步第2步优先工作"之前出现
2.worker只有一个实例，worker部分是否同时运行？

Answer 1

这个问题有点老了，但无论如何都会回答，因为我偶然发现了同样的东西。

我认为这只是因为你的电脑足够快，一旦它击中这个代码：

Source(1 to 10).map("job: " + _) ~> priorityPool1.jobsIn Source(1 to 10).map("priority job: " + _) ~> priorityPool1.priorityJobsIn

当它发送第二个10号时，第一个10号已经处理完毕。我认为，由于这个问题，他们将示例更改为100，但在我的计算机上，我仍然看到与您类似的结果，但如果您使用节流来减慢速度，您将看到您所期望的结果：

Source(1 to 10) .throttle(1, 0.1.second, 1, ThrottleMode.shaping) .map("job: " + _) ~> priorityPool1.jobsIn Source(1 to 10) .throttle(1, 0.1.second, 1, ThrottleMode.shaping) .map("priority job: " + _) ~> priorityPool1.priorityJobsIn

所以，这并不是说结果不正确，只是在并行处理中，你的计算机可能太快了。

当然，这里的节流仅用于减慢计算速度，并查看我们的学习示例是否有效，除非您确实想要放慢计算速度，否则不应在生产中使用节流。