उदाहरणों के साथ अतुल्यकालिक प्रोग्रामिंग: पुनर्निर्माण java.util.concurrent.CompletableFuture तरीके

यदि इस वर्ग के लिए स्रोत कोड खुला है तो पुनर्निर्माण क्यों आवश्यक है?

यदि केवल इसलिए कि हुड के तहत एक उच्च अनुकूलित, हार्ड-टू-रीड कोड है, जिसका अध्ययन थोड़ा शैक्षणिक रूप से देता है।

इसलिए, हम उनके विनिर्देशों के अनुसार संचालन के शब्दार्थों को फिर से बनाएंगे, और कार्यात्मक रूप से समकक्ष, समझने योग्य और पठनीय कोड लिखेंगे, हालांकि यह मेमोरी खपत और प्रोसेसर समय के संदर्भ में सबसे किफायती नहीं हो सकता है।

आइए अपेक्षाकृत सरल विधि से शुरुआत करें:

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) Returns a new CompletableFuture that is asynchronously completed by a task running in the given executor with the value obtained by calling the given Supplier. Type Parameters: U - the function's return type Parameters: supplier - a function returning the value to be used to complete the returned CompletableFuture executor - the executor to use for asynchronous execution Returns: the new CompletableFuture 

विनिर्देश को ध्यान से पढ़ें:

 Returns a new CompletableFuture 

यही है, टाइप CompletableFuture या उसके उपवर्ग का एक ऑब्जेक्ट बनाया जाता है और परिणामस्वरूप वापस आ जाता है।

 that is asynchronously completed by a task running in the given executor` 

इसके अलावा, एक कार्य निष्पादित किया जाता है जिसे Executor के ई पर निष्पादित किया जाता है।
जैसा कि हम जानते हैं, Executor केवल Runnable वस्तुओं को स्वीकार करता है।
रननेबल एक इंटरफ़ेस है, और पहली वस्तु इसे अच्छी तरह से लागू कर सकती है - इसलिए हम एक वस्तु में दो कार्यों को जोड़ देंगे।

  completed ... with the value obtained by calling the given Supplier. 

इस Runnable को प्रदत्त Supplier को कॉल करना चाहिए और प्राप्त मूल्य के साथ बनाए गए CompletableFuture पूरा करना चाहिए।

Supplier एक पैरामीटर रहित फ़ंक्शन है, इसलिए यह सब एन्कोडिंग बहुत सरल है:

  class CompletableFutureForSupplyAsync<U> extends CompletableFuture<U> implements Runnable { Supplier<U> supplier; public CompletableFutureForSupplyAsync(Supplier<U> supplier) { this.supplier = supplier; } public void run() { try { U result = supplier.get(); super.complete(result); } catch (Throwable e) { super.completeExceptionally(e); } } } public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) { CompletableFutureForSupplyAsync<U> task = new CompletableFutureForSupplyAsync<>(supplier); executor.execute(task); return task; } 

निम्नलिखित उदाहरण थोड़ा और अधिक जटिल है:

 public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) Returns a new CompletionStage that, when this stage completes normally, is executed using the supplied Executor, with this stage's result as the argument to the supplied function. See the CompletionStage documentation for rules covering exceptional completion. Specified by: thenApplyAsync in interface CompletionStage<T> Type Parameters: U - the function's return type Parameters: fn - the function to use to compute the value of the returned CompletionStage executor - the executor to use for asynchronous execution Returns: the new CompletionStage 

Returns a new CompletionStage that... is executed using the supplied Executor

यहां हमें सीधे Runnable के रूप में जारी करने के लिए बनाई गई वस्तु की व्यवस्था करने की पेशकश की जाती है।

... with this stage's result as the argument to the supplied function.

लेकिन यह पहले से अधिक दिलचस्प है। हमारे पास दिए गए फ़ंक्शन का एक पैरामीटर है, और इस पैरामीटर का मान वर्तमान CompletionStage पूरा करने वाला मान है। thenApplyAsync कॉल करने के समय thenApplyAsync यह मान ज्ञात नहीं हो सकता है, इसलिए हम तुरंत thenApplyAsync पर कार्य नहीं चला सकते हैं। इसके बजाय, हमें वर्तमान CompletionStage साथ बातचीत करनी चाहिए,
ताकि इसके पूरा होने के समय, यह अपने मूल्य को कार्य में स्थानांतरित कर दे। whenComplete के कई तरीकों में से एक है, जो इस उद्देश्य के लिए बिल्कुल उपयुक्त है, जब whenComplete :

 public CompletableFuture<T> whenComplete(BiConsumer<? super T,? super Throwable> action) Returns a new CompletionStage with the same result or exception as this stage, that executes the given action when this stage completes. 

यही है, नए बनाए गए कार्य ऑब्जेक्ट में, तर्क प्राप्त करने के लिए BiConsumer इंटरफ़ेस को लागू करना पर्याप्त है:

  class CompletableFutureForApplyAsync<T, U> extends CompletableFuture<U> implements Runnable, BiConsumer<T,Throwable> { Function<? super T,? extends U> fn; Executor executor; T arg; Throwable throwable; public CompletableFutureForApplyAsync(Function<? super T,? extends U> fn, Executor executor) { this.fn = fn; this.executor = executor; } @Override // implementation of BiConsumer interface public void accept(T argument, Throwable throwable) { if (throwable != null) { this.throwable = throwable; } else { this.arg = argument; } executor.execute(this); } @Override public void run() { if (throwable == null) { try { U result = fn.apply(arg); super.complete(result); } catch (Throwable e) { super.completeExceptionally(e); } } else { super.completeExceptionally(throwable); } } } public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor ) { CompletableFutureForApplyAsync<T,U> task = new CompletableFutureForApplyAsync<>(fn, executor); this.whenComplete(task); return task; } } 

अतुल्यकालिक प्रोग्रामिंग की प्रकृति को समझने के लिए यह उदाहरण बहुत महत्वपूर्ण है, इसलिए एक बार फिर हम इसके मुख्य चरणों को सूचीबद्ध करते हैं:

1) एक अतुल्यकालिक प्रक्रिया बनाई गई है:

  CompletableFutureForApplyAsync<T,U> task = new CompletableFutureForApplyAsync<>(fn, executor); 

2) यह अभी तक निष्पादन के लिए तैयार नहीं है, इसलिए हम लापता तर्क के प्रदाता से भविष्य में हमारे पास इस तर्क को पारित करने के लिए कहते हैं, हमने जो विधि प्रस्तुत की है, उसे कॉल करके:

  this.whenComplete(task); 

3) इस पद्धति में हम न केवल प्राप्त तर्क को बचाते हैं, बल्कि निष्पादन के लिए कार्य भी चलाते हैं ( accept () विधि देखें)।

4) कार्य का निष्पादन हमें दिए गए फ़ंक्शन की पूर्ति और परिणाम को बचाने के लिए कम हो गया है।
यह परिणाम ठीक उसी तरह हो सकता है जब हमारी नव-निर्मित वस्तु पर लागू होने वाली (जब) ​​विधि का उपयोग करके अन्य प्रक्रियाओं द्वारा अनुरोध किया जाता है, ताकि हम मनमाने ढंग से लंबाई की अतुल्यकालिक प्रक्रियाओं की एक श्रृंखला बना सकें। लेकिन यह श्रृंखला बिना किसी समानता के, सख्ती से क्रमिक रूप से पूरी होगी।

लेकिन समानांतर शाखाओं वाले अधिक जटिल गणना आरेख को कैसे चित्रित किया जाए?
ऐसा करने के लिए, thenCombineAsync विधि का उपयोग करें।

यदि पिछले उदाहरण में हमने एक तर्क के साथ अतुल्यकालिक प्रक्रिया शुरू की, तो इस में - दो के साथ।

इस मामले में, दोनों तर्कों की गणना समानांतर में हो सकती है।

 ublic <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor) Description copied from interface: CompletionStage Returns a new CompletionStage that, when this and the other given stage complete normally, is executed using the supplied executor, with the two results as arguments to the supplied function. 

यहां, सबकुछ thenApplyAsync उदाहरण में thenApplyAsync साथ thenApplyAsync , लेकिन फ़ंक्शन पैरामीटर में पहले से ही दो तर्क हैं, और thenApplyAsync CompletionStage<? extends U> other पैरामीटर जोड़ा गया है CompletionStage<? extends U> other CompletionStage<? extends U> other , जो दूसरे तर्क का अतुल्यकालिक प्रदाता है।

हम कैसे सुनिश्चित करते हैं कि दूसरा तर्क संसाधित किया गया है?

खैर, सबसे पहले, एक चर T arg बजाय T arg दो का वर्णन करें: T arg1; U arg2; T arg1; U arg2; , एक विधि के बजाय सार्वजनिक void accept(T argument, Throwable throwable) दो का वर्णन करते हैं - accept1 और accept2 ,
जिनमें से प्रत्येक अपने तर्क के साथ काम करता है।

उसी समय, निर्माणाधीन हमारी वस्तु अब BiConsumer<T,Throwable> लागू करती BiConsumer<T,Throwable> और हम अतुल्यकालिक गणना ग्राफ के नोड्स को जोड़ने के लिए एक महत्वपूर्ण वाक्य नहीं लिख सकते हैं

  this.whenComplete(task); 

सौभाग्य से, एक कार्यात्मक इंटरफ़ेस ऑब्जेक्ट को एक अलग वर्ग में निष्कर्ष निकाले बिना, विधि के संदर्भ में दर्शाया जा सकता है:

  this.whenComplete(task::accept1); other.whenComplete(task::accept2); 

यही है, वर्तमान this वस्तु पहले तर्क की आपूर्ति करती है, और other वस्तु other वस्तु।

यहां केवल उन विधियों के कोड हैं जिन्हें बदलना होगा ताकि वे अपने तर्क के आने पर तुरंत कार्य शुरू न करें, लेकिन दूसरे की प्राप्ति की भी जांच करें:

  public synchronized void accept1(T argument, Throwable throwable) { if (throwable != null) { this.throwable = throwable; executor.execute(this); } else { this.arg1 = argument; if (arg2 != null) { executor.execute(this); } } } 

स्वीकार 2 विधि इसी तरह वर्णित है।

ध्यान दें:

• तरीके सिंक्रनाइज़ हो जाते हैं (हम सामान्य डेटा के साथ काम करते हैं)
• त्रुटि संचरण के मामले में, दूसरे तर्क की प्रतीक्षा करना आवश्यक नहीं है।
• यह जाँचना कि तर्क null तुलना में आता है सबसे अच्छा तरीका नहीं है, शायद आपको प्रत्येक तर्क के लिए बूलियन चर जोड़ने की आवश्यकता है।

इस तरह, आप दो की तुलना में बड़ी संख्या में तर्कों से अतुल्यकालिक प्रक्रियाएं बना सकते हैं, लेकिन विचार तुरंत सामने आता है - क्या आप अभी भी मापदंडों के लिए एक अलग वर्ग बना सकते हैं, इसलिए प्रत्येक पैरामीटर प्राप्त करने के लिए अपनी खुद की विधि लिखने के लिए नहीं, लेकिन क्या यह गतिशील रूप से पैरामीटर बना रहा है?

  Parameter<Integer> arg1 = new Parameter<>(); Parameter<Float> arg2 = new Parameter<>(); ... future1.whenComplete(arg1); future2.whenComplete(arg2); 

हां, आप इस तरह की कक्षा बना सकते हैं, लेकिन अगली बार उस पर अधिक।

पूर्वगामी का एक संक्षिप्त सारांश:

• एक एसिंक्रोनस प्रोग्राम इंटरकनेक्टेड एसिंक्रोनस प्रक्रियाओं का एक नेटवर्क है,
  जैसे एक बहु-थ्रेडेड प्रोग्राम इंटरकनेक्टेड थ्रेड थ्रेड्स (थ्रेड्स) का एक नेटवर्क है।

लेकिन संचार प्रवाह और अतुल्यकालिक प्रक्रियाओं के साधन मौलिक रूप से भिन्न हैं।

धाराएँ सेमाफोर, अवरोधक कतारों और अन्य समान वस्तुओं का उपयोग करके जुड़ी होती हैं,
जो प्राप्तकर्ता स्ट्रीम को ब्लॉक कर देता है अगर जानकारी अभी तक नहीं आई है, लेकिन स्ट्रीम पहले से ही पुल-आधारित ऑपरेशन का उपयोग करके इसे निकालने की कोशिश कर रहा है।

अतुल्यकालिक प्रक्रियाएं - प्राप्तकर्ता केवल तब तक निष्पादन नहीं करते हैं जब तक कि उन्हें आवश्यक सभी जानकारी तैयार न हो जाए।
वे तब तक प्रतीक्षा करते हैं जब तक कि सूचना प्रदाता खुद इसे एक पुश-आधारित ऑपरेशन के माध्यम से पारित न करें।
इसके कारण, वे प्रतीक्षा करते समय स्टैक पर मेमोरी खर्च नहीं करते हैं, और इसलिए, निष्पादन के थ्रेड्स की तुलना में बहुत कम मेमोरी लेते हैं।

• अतुल्यकालिक प्रक्रियाओं के एक नेटवर्क का निर्माण वस्तुओं को बनाने और उन्हें एक साथ जोड़ने के लिए कम किया जाता है, और अधिक सटीक रूप से, उनके उप-विषयों को जोड़ने के लिए - पैरामीटर: प्राप्तकर्ता के इनपुट पैरामीटर का पता सूचना प्रदाता के आउटपुट पैरामीटर को प्रेषित किया जाता है।

CompletableFuture तरीकों का सेट ठीक यही करता है, और सिद्धांत रूप में, आप इन तरीकों के बिना वस्तुओं को स्पष्ट रूप से बनाकर कर सकते हैं, जैसा कि ऊपर दिए गए उदाहरणों में दिखाया गया है।
लेकिन इसके लिए उन वर्गों के समान वर्ग होना आवश्यक है जिन्हें इन उदाहरणों में वर्णित किया गया था।
किसी कारण से, java.util.concurrent के रचनाकारों ने उपयोगकर्ताओं को इन कक्षाओं तक पहुंच नहीं देने और उन्हें CompletableFuture कोड की गहराई में CompletableFuture के लिए चुना।

जो लोग बनाए जा रहे अतुल्यकालिक नेटवर्क का दृश्य प्रतिनिधित्व करना चाहते हैं वे उपरोक्त उदाहरणों को जारी रखते हुए इन वर्गों को फिर से संगठित कर सकते हैं। उदाहरणों के लिए स्रोत कोड Github पर उपलब्ध है ।