У меня есть асинхронный поток управления как следующее:
ActorA ! DoA(dataA, callback1, callbackOnErrorA)
def callback1() = {
...
ActorB ! DoB(dataB, callback2, callbackOnErrorB)
}
def callback2() = {
ActorC ! DoC(dataC, callback3, callbackOnErrorC)
}
...
Как я разделил бы этот поток на несколько частей (продолжения) и последовательно отправил бы их различным агентам (или потоки/задачи) при поддержании полного состояния?
Любая ценившая подсказка, Спасибо
Это очень упрощено, но показывает, как разделить единый поток управления между тремя участниками, передавая состояние каждому:
package blevins.example
import scala.continuations._
import scala.continuations.ControlContext._
import scala.actors.Actor._
import scala.actors._
object App extends Application {
val actorA, actorB, actorC = actor {
receive {
case f: Function1[Unit,Unit] => { f() }
}
}
def handle(a: Actor) = shift { k: (Unit=>Unit) =>
a ! k
}
// Control flow to split up
reset {
// this is not handled by any actor
var x = 1
println("a: " + x)
handle(actorA) // actorA handles the below
x += 4
println("b: " + x)
handle(actorB) // then, actorB handles the rest
var y = 2
x += 2
println("c: " + x)
handle(actorC) // and so on...
y += 1
println("d: " + x + ":" + y)
}
}
Мне нравится использовать scalaz.concurrent.Promise
. Этот пример не совсем похож на тот, что в вашем вопросе, но он дает вам представление.
object Async extends Application {
import scalaz._
import Scalaz._
import concurrent._
import concurrent.strategy._
import java.util.concurrent.{ExecutorService, Executors}
case class ResultA(resultb: ResultB, resulta: ResultC)
case class ResultB()
case class ResultC()
run
def run {
implicit val executor: ExecutorService = Executors.newFixedThreadPool(8)
import Executor.strategy
val promiseA = doA
println("waiting for results")
val a: ResultA = promiseA.get
println("got " + a)
executor.shutdown
}
def doA(implicit s: Strategy[Unit]): Promise[ResultA] = {
println("triggered A")
val b = doB
val c = doC
for {bb <- b; cc <- c} yield ResultA(bb, cc)
}
def doB(implicit s: Strategy[Unit]): Promise[ResultB] = {
println("triggered B")
promise { Thread.sleep(1000); println("returning B"); ResultB() }
}
def doC(implicit s: Strategy[Unit]): Promise[ResultC] = {
println("triggered C")
promise { Thread.sleep(1000); println("returning C"); ResultC() }
}
}
Вывод:
triggered A
triggered B
triggered C
waiting for results
returning B
returning C
got ResultA(ResultB(),ResultC())
Введение в Scalaz параллелизм вы найдете в этой презентации от Runar.
Этот подход не такой гибкий, как Actors, но лучше компонуется и не может зайти в тупик.