13

次のような非同期制御フローがあります。

ActorA ! DoA(dataA, callback1, callbackOnErrorA)

def callback1() = {
  ...
  ActorB ! DoB(dataB, callback2, callbackOnErrorB)
}

def callback2() = {
  ActorC ! DoC(dataC, callback3, callbackOnErrorC)
} 

...

このフローをいくつかの部分（継続）に分割し、全体的な状態を維持しながら、これらをさまざまなアクター（またはスレッド/タスク）に順番にディスパッチするにはどうすればよいですか？

ヒントをいただければ幸いです、ありがとう

4

3 に答える 3

9

私は使用するのが好き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())

Runarのこのプレゼンテーションでは、 Scalazの同時実行性の概要を説明しています。

このアプローチはアクターほど柔軟ではありませんが、構成が優れており、デッドロックすることはありません。

于 2010-03-15T14:05:35.787 に答える
7

これは非常に単純化されていますが、単一の制御フローを3つのアクターに分割し、それぞれに状態を渡す方法を示しています。

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)
  }

}
于 2010-03-15T22:56:27.770 に答える
1

Akkaの先物とそれらの構成方法またはscalazの約束を参照してください。これらはほぼ同じで、わずかな違いしかありません。

于 2011-09-07T18:26:52.287 に答える