निरंतरता का उपयोग करके एक एसिंक नियंत्रण-प्रवाह कैसे विभाजित और प्रेषित करें?

Question

मैं की तरह एक अतुल्यकालिक नियंत्रण प्रवाह है निम्नलिखित:

ActorA ! DoA(dataA, callback1, callbackOnErrorA) 

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

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

... 

मैं कैसे कई भागों (निरंतरता) में इस प्रवाह को विभाजित होता है और क्रमिक रूप से अलग अलग अभिनेताओं (या धागे/कार्य), जबकि समग्र स्थिति को बनाए रखने के लिए इन प्रेषण ?

सराहना की किसी भी संकेत है, धन्यवाद

Answer 1

यह बहुत सरल है, लेकिन पता चलता है कि तीन कलाकारों के बीच एक भी नियंत्रण प्रवाह को विभाजित करने के लिए, प्रत्येक के साथ राज्य गुजर:

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

} 

Answer 2

मैं 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()) 

आप इस presentation रुनर से में Scalaz संगामिति का परिचय मिल जाएगा।

यह दृष्टिकोण अभिनेताओं के रूप में लचीला नहीं है, लेकिन बेहतर बनाता है और डेडलॉक नहीं कर सकता है।

Answer 3

देखें Akka's Futures and how to compose them या scalaz's Promises, वे लगभग समान हैं, देखते हैं केवल मामूली मतभेद।