स्कैला भविष्य अनुक्रम और टाइमआउट हैंडलिंग

Question

कुछ अच्छे संकेत हैं कि वायदा with timeouts कैसे गठबंधन करें। हालांकि मैं कैसे भविष्य अनुक्रम sequenceOfFutures

मेरा पहला दृष्टिकोण यह

import scala.concurrent._ 
import scala.concurrent.duration._ 
import scala.concurrent.ExecutionContext.Implicits._ 

object FutureSequenceScala extends App { 
    println("Creating futureList") 

    val timeout = 2 seconds 
    val futures = List(1000, 1500, 1200, 800, 2000) map { ms => 
    val f = future { 
     Thread sleep ms 
     ms toString 
    } 
    Future firstCompletedOf Seq(f, fallback(timeout)) 
    } 

    println("Creating waitinglist") 
    val waitingList = Future sequence futures 
    println("Created") 

    val results = Await result (waitingList, timeout * futures.size) 
    println(results) 

    def fallback(timeout: Duration) = future { 
    Thread sleep (timeout toMillis) 
    "-1" 
    } 
} 

तरह लग रहा है वहाँ वायदा के एक क्रम में समय समाप्ति को संभालने के लिए एक बेहतर तरीका है के साथ ऐसा करना उत्सुक हूँ या यह एक है वैध समाधान?

Answer 1

यहां आपके कोड में कुछ चीज़ें हैं जिन्हें आप पुनर्विचार करना चाहेंगे। शुरुआत करने वालों के लिए, मैं ExecutionContext में कार्यों को सबमिट करने का एक बड़ा प्रशंसक नहीं हूं, जिसमें टाइमआउट अनुकरण करने का एकमात्र उद्देश्य है और उनमें Thread.sleep भी उपयोग किया जाता है। sleep कॉल अवरुद्ध हो रहा है और आप शायद निष्पादन संदर्भ में कोई कार्य करने से बचना चाहते हैं जो निश्चित रूप से निश्चित समय की प्रतीक्षा करने के लिए अवरुद्ध है। मैं अपने उत्तर here से चोरी करने जा रहा हूं और सुझाव देता हूं कि शुद्ध टाइमआउट हैंडलिंग के लिए, आपको उस उत्तर में उल्लिखित कुछ का उपयोग करना चाहिए। HashedWheelTimer एक बेहद कुशल टाइमर कार्यान्वयन है जो कि बस सोता है कि एक काम की तुलना में समय समाप्ति के लिए बेहतर अनुकूल है।

अब, यदि आप उस मार्ग पर जाते हैं, तो अगला परिवर्तन मैं प्रत्येक भविष्य के लिए व्यक्तिगत टाइमआउट से संबंधित असफलताओं को संभालने वाली चिंताओं का सुझाव दूंगा। यदि आप पूरी तरह विफल होने की व्यक्तिगत विफलता चाहते हैं तो Futuresequence कॉल से वापस लौटा, तो कुछ भी अतिरिक्त नहीं करें। यदि आप नहीं चाहते हैं कि ऐसा करने के लिए, और बदले चाहते कुछ डिफ़ॉल्ट मान के बजाय वापस जाने के लिए समय समाप्त, तो आप Future इस तरह पर recover उपयोग कर सकते हैं:

withTimeout(someFuture).recover{ 
    case ex:TimeoutException => someDefaultValue 
} 

आपके द्वारा किए गए एक बार, आप ले जा सकते हैं इस तरह गैर-अवरुद्ध कॉलबैक का लाभ और कर कुछ:

waitingList onComplete{ 
    case Success(results) => //handle success 
    case Failure(ex) => //handle fail 
} 

प्रत्येक भविष्य समय समाप्त है और इस प्रकार सिर्फ असीम नहीं चलेंगे। वहां ब्लॉक करने के लिए आईएमओ की आवश्यकता नहीं है और atMost पैरामीटर Await.result के माध्यम से टाइमआउट हैंडलिंग की एक अतिरिक्त परत प्रदान करें। लेकिन मुझे लगता है कि यह मानता है कि आप गैर-अवरुद्ध दृष्टिकोण के साथ ठीक हैं। यदि आपको वास्तव में वहां अवरुद्ध करने की आवश्यकता है, तो आपको timeout * futures.size समय की प्रतीक्षा नहीं करनी चाहिए। ये वायदा समानांतर में चल रहे हैं; समय-समय पर केवल वायदा के लिए व्यक्तिगत टाइमआउट (या सीपीयू/समय में किसी भी देरी के लिए खाते में थोड़ी देर तक) होने की आवश्यकता होनी चाहिए। यह निश्चित रूप से टाइमआउट * वायदा की कुल संख्या नहीं होनी चाहिए।

Answer 2

यहां एक संस्करण है जो दिखाता है कि fallback आपकी अवरोध कितनी खराब है।

ध्यान दें कि निष्पादक एकल धागा है और आप कई फॉलबैक बना रहे हैं।

@cmbaxter सही है, आपका मास्टर टाइमआउट timeout * futures.size नहीं होना चाहिए, यह बड़ा होना चाहिए!

@cmbaxter भी सही है कि आप गैर-अवरोधन करना चाहते हैं। एक बार ऐसा करने के बाद, और आप टाइमआउट लागू करना चाहते हैं, तो आप इसके लिए टाइमर घटक चुनेंगे, उसके लिंक किए गए उत्तर (आपके लिंक किए गए उत्तर से भी लिंक) देखें।

उस ने कहा, मुझे अभी भी my answer from your link पसंद है, जहां तक ​​लूप में बैठे हुए अगली चीज़ के लिए इंतजार कर रहे हैं, जो कि टाइमआउट वास्तव में सरल है।

यह सिर्फ वायदा और उनके टाइमआउट और फ़ॉलबैक मान की एक सूची लेता है।

हो सकता है कि इस तरह के एक सरल अनुप्रयोग के रूप में उस के लिए एक उपयोग के मामले, है कि अभी कुछ परिणाम (अपने परीक्षण की तरह) के लिए ब्लॉक और से पहले नहीं बाहर निकलने के परिणामों में हैं

import scala.concurrent._ 
import scala.concurrent.duration._ 
import scala.concurrent.ExecutionContext 

import java.util.concurrent.Executors 
import java.lang.System.{ nanoTime => now } 

object Test extends App { 
    //implicit val xc = ExecutionContext.global 
    implicit val xc = ExecutionContext fromExecutorService (Executors.newSingleThreadExecutor) 

    def timed[A](body: =>A): A = { 
    val start = now 
    val res = body 
    val end = now 
    Console println (Duration fromNanos end-start).toMillis + " " + res 
    res 
    } 
    println("Creating futureList") 

    val timeout = 1500 millis 
    val futures = List(1000, 1500, 1200, 800, 2000) map { ms => 
    val f = future { 
     timed { 
     blocking(Thread sleep ms) 
     ms toString 
     } 
    } 
    Future firstCompletedOf Seq(f, fallback(timeout)) 
    } 

    println("Creating waitinglist") 
    val waitingList = Future sequence futures 
    println("Created") 

    timed { 
    val results = Await result (waitingList, 2 * timeout * futures.size) 
    println(results) 
    }  
    xc.shutdown 

    def fallback(timeout: Duration) = future { 
    timed { 
     blocking(Thread sleep (timeout toMillis)) 
     "-1" 
    } 
    } 
} 

क्या हुआ:।

Creating futureList 
Creating waitinglist 
Created 
1001 1000 
1500 -1 
1500 1500 
1500 -1 
1200 1200 
1500 -1 
800 800 
1500 -1 
2000 2000 
1500 -1 
List(1000, 1500, 1200, 800, 2000) 
14007()