जावा / GPars आधारित अभिनेताओं का परिचय, भाग I

GPars पुस्तकालय का एपीआई और मध्यम जटिलता के एक बहु- स्तरीय कार्य का समाधान, जिसके परिणाम "राष्ट्रीय अर्थव्यवस्था" में उपयोगी हो सकते हैं, संक्षेप में माना जाता है।

यह लेख "जावा में मल्टीकोर प्रोग्रामिंग" पाठ्यक्रम को पढ़ने की तैयारी में जावा प्रोग्रामर के लिए उपलब्ध विभिन्न अभिनेता पुस्तकालयों के एक अध्ययन के दौरान लिखा गया था।

मैं ऑनलाइन शिक्षा प्लेटफॉर्म udemy.com (कौरसेरा / एडएक्स के समान) पर जावा डेवलपर्स कोर्स के लिए स्काला भी सिखाता हूं।

यह लेखों की श्रृंखला में पहला लेख है जिसका उद्देश्य कुछ मॉडल समस्या पर अन्य पुस्तकालयों में कार्यान्वयन के साथ एपीआई, प्रदर्शन, और अक्का अभिनेताओं की तुलना करना है। यह लेख GPars पर इस तरह की समस्या और समाधान प्रदान करता है।

GPars एक लाइब्रेरी है जो क्लॉजुर के लिए विभिन्न समानांतर कंप्यूटिंग दृष्टिकोणों के लिए व्यापक समर्थन के साथ लिखा गया है।
GPars पेशेवरों

• स्रोत कोड जावा में लिखा गया है (स्काला में लिखित अक्का के विपरीत)। "देशी" प्रोग्रामिंग भाषा में "हुड के नीचे क्या है" यह देखना हमेशा दिलचस्प होता है
• GPars दृष्टिकोणों का एक पूरा "चिड़ियाघर" है (अभिनेता, एजेंट, एसटीएम, सीएसपी, डेटाफ़्लो)
• GPars जावा में लिखित क्लोजर रनटाइम लाइब्रेरी से कक्षाओं का उपयोग करता है। रुमेज को दिलचस्प

"स्थापना" GPars

मावेन GPars और Groovy में प्लग करें

<dependency> <groupId>org.codehaus.gpars</groupId> <artifactId>gpars</artifactId> <version>1.1.0</version> </dependency> <dependency> <groupId>org.codehaus.groovy</groupId> <artifactId>groovy-all</artifactId> <version>2.2.2</version> </dependency> 

मावेन के बिना, रिपॉजिटरी से केवल GPars-1.1.0 ( स्रोत ) और ग्रूवी -२.२ ( स्रोत ) डाउनलोड करें और प्रोजेक्ट से कनेक्ट करें।

स्टेटलेस अभिनेता

सरल उदाहरणों के साथ शुरू करते हैं।
हम अभिनेता को संदेश भेज रहे हैं।

 import groovyx.gpars.actor.*; public class Demo { public static void main(String[] args) throws Exception { Actor actor = new DynamicDispatchActor() { public void onMessage(String msg) { System.out.println("receive: " + msg); } }.start(); actor.send("Hello!"); System.in.read(); } } >> receive: Hello! 

संदेश भेजें और प्रतिक्रिया की प्रतीक्षा करें

 import groovyx.gpars.actor.*; public class Demo { public static void main(String[] args) throws Exception { Actor actor = new DynamicDispatchActor() { public void onMessage(String msg) { System.out.println("ping: " + msg); getSender().send(msg.toUpperCase()); } }.start(); System.out.println("pong: " + actor.sendAndWait("Hello!")); } } >> ping: Hello! >> pong: HELLO! 

हम एक संदेश भेजते हैं और एक अतुल्यकालिक कॉलबैक को लटकाते हैं

 import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.*; public class Demo { public static void main(String[] args) throws Exception { Actor actor = new DynamicDispatchActor() { public void onMessage(String msg) { System.out.println("ping: " + msg); getSender().send(msg.toUpperCase()); } }.start(); actor.sendAndContinue("Hello!", new MessagingRunnable<String>() { protected void doRun(String msg) { System.out.println("pong: " + msg); } }); System.in.read(); } } >> ping: Hello! >> pong: HELLO! 

प्राप्त संदेश के प्रकार से पैटर्न का मिलान करना

 import groovyx.gpars.actor.*; public class Demo { public static void main(String[] args) throws Exception { Actor actor = new DynamicDispatchActor() { public void onMessage(String arg) { getSender().send(arg.toUpperCase()); } public void onMessage(Long arg) { getSender().send(1000 + arg); } }.start(); System.out.println("42.0 -> " + actor.sendAndWait(42.0)); } } >> Hello! -> HELLO! >> 42 -> 1042 

पैटर्न मिलान से उपयुक्त हैंडलर नहीं मिला

 import groovyx.gpars.actor.*; public class Demo { public static void main(String[] args) throws Exception { Actor actor = new DynamicDispatchActor() { public void onMessage(String arg) { getSender().send(arg.toUpperCase()); } public void onMessage(Long arg) { getSender().send(1000 + arg); } }.start(); System.out.println("42.0 -> " + actor.sendAndWait(42.0)); } } >> An exception occurred in the Actor thread Actor Thread 1 >> groovy.lang.MissingMethodException: No signature of method: >> net.golovach.Demo_4$1.onMessage() is applicable for argument types: (java.lang.Double) values: [42.0] >> Possible solutions: onMessage(java.lang.Long), onMessage(java.lang.String) >> at org.codehaus.groovy.runtime.ScriptBytecodeAdapter ... >> ... 

जो दिख रहा है
- "पैटर्न मिलान" ऑनमैसेज के एक उपयुक्त अतिभारित संस्करण का चयन करता है (<एक- arg>) विधि, अगर कोई नहीं है, तो हमें एक अपवाद मिलता है
- अभिनेता "डेमॉन" थ्रेड्स के एक पूल के आधार पर काम करते हैं, इसलिए हमें किसी तरह जेवीएम को समय से पहले समाप्त करने से रोकने के लिए मुख्य () विधि (मैंने System.in.read () का इस्तेमाल किया) के संचालन को स्थगित करने की आवश्यकता है
- उत्तर () विधि के उदाहरण पर, हम देखते हैं कि जब डायनेमिकडिस्पैक्टर से विरासत में मिला है, तो बहुत सारी विधियां अभिनेता के "नेमस्पेस" में आती हैं (उत्तर, replIfExists, getSender, समाप्त, ...

हालाँकि GPars के लेखक डायनामेडस्पाइचैक्टरएक्टर स्टेट स्टेटलेस एक्टर के वारिस कहलाते हैं, ये जावा क्लासेस के सामान्य उदाहरण हैं जो उत्परिवर्ती क्षेत्र हो सकते हैं और उनमें अपना राज्य संग्रहीत कर सकते हैं। इसका प्रदर्शन करें

 import groovyx.gpars.actor.*; import java.util.ArrayList; import java.util.List; public class StatelessActorTest { public static void main(String[] args) throws InterruptedException { Actor actor = new DynamicDispatchActor() { private final List<Double> state = new ArrayList<>(); public void onMessage(final Double msg) { state.add(msg); reply(state); } }.start(); System.out.println("answer: " + actor.sendAndWait(1.0)); System.out.println("answer: " + actor.sendAndWait(2.0)); System.out.println("answer: " + actor.sendAndWait(3.0)); System.out.println("answer: " + actor.sendAndWait(4.0)); System.out.println("answer: " + actor.sendAndWait(5.0)); } } >> answer: [1.0] >> answer: [1.0, 2.0] >> answer: [1.0, 2.0, 3.0] >> answer: [1.0, 2.0, 3.0, 4.0] >> answer: [1.0, 2.0, 3.0, 4.0, 5.0] 

स्टेटफुल एक्टर

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

 import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.*; import static java.util.Arrays.asList; public class StatefulActorTest { public static void main(String[] args) throws Exception { Actor actor = new MyStatefulActor().start(); actor.send("A"); actor.send(1.0); actor.send(Arrays.asList(1, 2, 3)); actor.send("B"); actor.send(2.0); actor.send(Arrays.asList(4, 5, 6)); System.in.read(); } private static class MyStatefulActor extends DefaultActor { protected void act() { loop(new Runnable() { public void run() { react(new MessagingRunnable<Object>(this) { protected void doRun(final Object msg) { System.out.println("react: " + msg); } }); } }); } } } >> react: A >> react: 1.0 >> react: [1, 2, 3] >> react: B >> react: 2.0 >> react: [4, 5, 6] 

हालाँकि, राज्य टेम्पलेट का वादा किया गया क्रियान्वयन बिल्कुल भी बदबूदार नहीं है। इस तरह से चलते हैं (Java ऐसी ट्रिक्स के लिए सबसे अच्छी भाषा नहीं है, Clojure / Scala पर यह कोड बहुत अधिक कॉम्पैक्ट दिखता है)

 import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.*; import java.util.List; import static java.util.Arrays.asList; public class StatefulActorTest { public static void main(String[] args) throws Exception { Actor actor = new MyStatefulActor().start(); actor.send("A"); actor.send(1.0); actor.send(asList(1, 2, 3)); actor.send("B"); actor.send(2.0); actor.send(asList(4, 5, 6)); System.in.read(); } private static class MyStatefulActor extends DefaultActor { protected void act() { loop(new Runnable() { public void run() { react(new MessagingRunnable<String>(this) { protected void doRun(final String msg) { System.out.println("Stage #0: " + msg); react(new MessagingRunnable<Double>() { protected void doRun(final Double msg) { System.out.println(" Stage #1: " + msg); react(new MessagingRunnable<List<Integer>>() { protected void doRun(final List<Integer> msg) { System.out.println(" Stage #2: " + msg + "\n"); } }); } }); } }); } }); } } } >> Stage #0: A >> Stage #1: 1.0 >> Stage #2: [1, 2, 3] >> >> Stage #0: B >> Stage #1: 2.0 >> Stage #2: [4, 5, 6] 

खैर, या चलो अनाम वर्गों के इस भयानक घोंसले के शिकार से छुटकारा पाएं और "राज्य को भौतिक बनाएं"

 import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.*; import java.util.List; import static java.util.Arrays.asList; public class StatefulActorTest { public static void main(String[] args) throws Exception { Actor actor = new MyStatefulActor().start(); actor.send("A"); actor.send(1.0); actor.send(asList(1, 2, 3)); actor.send("B"); actor.send(2.0); actor.send(asList(4, 5, 6)); System.in.read(); } private static class MyStatefulActor extends DefaultActor { protected void act() { loop(new Runnable() { public void run() { react(new Stage0(MyStatefulActor.this)); } }); } } private static class Stage0 extends MessagingRunnable<String> { private final DefaultActor owner; private Stage0(DefaultActor owner) {this.owner = owner;} protected void doRun(final String msg) { System.out.println("Stage #0: " + msg); owner.react(new Stage1(owner)); } } private static class Stage1 extends MessagingRunnable<Double> { private final DefaultActor owner; private Stage1(DefaultActor owner) {this.owner = owner;} protected void doRun(final Double msg) { System.out.println(" Stage #1: " + msg); owner.react(new Stage2()); } } private static class Stage2 extends MessagingRunnable<List<Integer>> { protected void doRun(final List<Integer> msg) { System.out.println(" Stage #2: " + msg + "\n"); } } } 

हां, हां, मैं आपसे पूरी तरह सहमत हूं, जावा एक अत्यंत क्रियात्मक भाषा है।

यहां बताया गया है कि संक्रमण आरेख कैसा दिखता है (हमने तर्क को कांटा नहीं किया)

 // START // ----- // | // | // | // | +--------+ // +->| Stage0 | ---String----+ // +--------+ | // ^ v // | +--------+ // | | Stage1 | // List<Integer> +--------+ // | | // | +--------+ Double // +--| Stage2 |<-------+ // +--------+ 

घड़ी

मेरी समस्या को हल करने के लिए, मुझे एक टाइमर की आवश्यकता होगी - कुछ ऐसा जो मुझे निश्चित अवधि के अंत में सूचित करने के लिए प्रोग्राम किया जा सकता है। "नियमित" जावा में, हम java.util.concurrent.ScheduledThreadPoolExecutor या java.util.Timer का उपयोग सबसे कम करते हैं। लेकिन हम अभिनेताओं की दुनिया में हैं!
यह एक स्टेटफुल एक्टर है जो टाइमआउट के साथ प्रतिक्रिया () विधि में एक संदेश की प्रतीक्षा कर रहा है। यदि इस अवधि के दौरान कोई संदेश नहीं आता है, तो GPars अवसंरचना हमें एक अभिनेता भेजती है। संदेश के बारे में। (यह सिर्फ "टाइमटाइम" लाइन है) और हम अपने निर्माता को मध्यांतर निर्माणकर्ता से संदेश "वापस" करते हैं। यदि आप टाइमर को "बंद" करना चाहते हैं - तो उसे कोई अन्य संदेश भेजें (मैं उसे "KILL" स्ट्रिंग भेजूंगा)

 import groovyx.gpars.MessagingRunnable; import groovyx.gpars.actor.*; import groovyx.gpars.actor.impl.MessageStream; import static java.util.concurrent.TimeUnit.MILLISECONDS; public class Timer<T> extends DefaultActor { private final long timeout; private final T timeoutMsg; private final MessageStream replyTo; public Timer(long timeout, T timeoutMsg, MessageStream replyTo) { this.timeout = timeout; this.timeoutMsg = timeoutMsg; this.replyTo = replyTo; } protected void act() { loop(new Runnable() { public void run() { react(timeout, MILLISECONDS, new MessagingRunnable() { protected void doRun(Object argument) { if (Actor.TIMEOUT.equals(argument)) { replyTo.send(timeoutMsg); } terminate(); } }); } }); } } 

टाइमर का उपयोग करने का एक उदाहरण।
मैं दो टाइमर टाइमर और टाइमर बना रहा हूं, जो 1000 मी की देरी के साथ मुझे क्रमशः "एक्स" और "वाई" संदेश भेजेगा। लेकिन 500 मी के बाद मैंने अपना दिमाग बदल दिया और "नेक्ड" टाइमरएक्स।

 import groovyx.gpars.actor.Actor; import groovyx.gpars.actor.impl.MessageStream; public class TimerDemo { public static void main(String[] args) throws Exception { Actor timerX = new Timer<>(1000, "X", new MessageStream() { public MessageStream send(Object msg) { System.out.println("timerX send timeout message: '" + msg + "'"); return this; } }).start(); Actor timerY = new Timer<>(1000, "Y", new MessageStream() { public MessageStream send(Object msg) { System.out.println("timerY send timeout message: '" + msg + "'"); return this; } }).start(); Thread.sleep(500); timerX.send("KILL"); System.in.read(); } } >> timerY send timeout message: 'Y' 

समस्या बयान और समाधान योजना

निम्नलिखित बहुत सामान्य समस्या पर विचार करें।
1. हमारे पास कई सूत्र हैं जो अक्सर किसी न किसी फ़ंक्शन का कारण बनते हैं।
2. इस फ़ंक्शन के दो विकल्प हैं: किसी एकल तर्क को संसाधित करना और तर्कों की सूची को संसाधित करना।
3. यह फ़ंक्शन ऐसा है कि एक तर्क सूची को संसाधित करना प्रत्येक व्यक्तिगत प्रसंस्करण के योग से कम सिस्टम संसाधनों की खपत करता है।
4. कार्य प्रवाह और फ़ंक्शन के बीच कुछ बैचर लगाने का है, जो प्रवाह से तर्कों को एक "बंडल" में एकत्रित करता है, इसे फ़ंक्शन में भेजता है, यह सूची को संसाधित करता है, बैचर "प्रेषित" करता है और प्रेषक को परिणाम प्रदान करता है।
5. बैचर दो मामलों में तर्कों की एक सूची से गुजरता है: उन्होंने पर्याप्त आकार का एक "बंडल" एकत्र किया या एक समय-आउट अवधि के दौरान जिसके दौरान एक पूरा "बंडल" एकत्र करना संभव नहीं था, लेकिन थ्रेड्स के लिए परिणाम वापस करने का समय है।

आइए एक समाधान योजना देखें।
टाइमआउट 100ms, "बंडल" का अधिकतम आकार - 3 तर्क

समय 0 पर, प्रवाह T-0 तर्क "A" भेजता है। बैचर "साफ" स्थिति में है, पीढ़ी 0

 //time:0 // // T-0 --"A"-----> +-------+ generationId=0 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] 

एक पल के बाद, बैचर जानता है कि "ए" को छोटा करना और टी-0 पर वापस करना आवश्यक है। टाइमर पीढ़ी 0 के लिए शुरू किया

 // +-----+ timeoutMsg=0 // |Timer| timeout=100 //time:0.001 +-----+ // // T-0 +-------+ generationId=0 // T-1 |Batcher| argList=["A"] // T-2 +-------+ replyToList=[T-0] 

25 मिलीसेकंड के समय, टी -1 स्ट्रीम प्रसंस्करण के लिए "बी" भेजता है

 // +-----+ timeoutMsg=0 // |Timer| timeout=100 //time:25 +-----+ // // T-0 +-------+ generationId=0 // T-1 ---"B"----> |Batcher| argList=["A"] // T-2 +-------+ replyToList=[T-0] 

एक पल के बाद, बैचर जानता है कि "ए" और "बी" को छोटा करना और प्रवाह को टी-0 और टी -1 में वापस करना आवश्यक है

 // +-----+ timeoutMsg=0 // |Timer| timeout=100 //time:25.001 +-----+ // // T-0 +-------+ generationId=0 // T-1 |Batcher| argList=["A","B"] // T-2 +-------+ replyToList=[T-0,T-1] 

50 मिलीसेकंड के समय में, टी -2 स्ट्रीम प्रसंस्करण के लिए "सी" भेजता है

 // +-----+ timeoutMsg=0 // |Timer| timeout=100 //time:50 +-----+ // // T-0 +-------+ generationId=0 // T-1 |Batcher| argList=["A","B"] // T-2 ----"C"---> +-------+ replyToList=[T-0,T-1] 

एक पल के बाद, बैचर जानता है कि "ए", "बी" और "सी" की गणना करना आवश्यक है और इसे प्रवाह टी -0, टी -1 और टी -2 में लौटा देना चाहिए। पता चलता है कि "बंडल" भरा हुआ है और टाइमर "मारता है"

 // +-----+ timeoutMsg=0 // +-"KILL"->|Timer| timeout=100 //time:50.001 | +-----+ // | // T-0 +-------+ generationId=0 // T-1 |Batcher| argList=["A","B","C"] // T-2 +-------+ replyToList=[T-0,T-1,T-2] 

एक पल के बाद, बैचर एक अलग अभिनेता (एनोनिमस) को गणना के लिए डेटा देता है, राज्य को साफ करता है और पीढ़ी को 0 से 1 तक बदलता है

 //time:50.002 // // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] // // +---------+ argList=["A","B","C"] // |anonymous| replyToList=[T-0,T-1,T-2] // +---------+ 

एक पल के बाद ("स्टोरीबोर्ड" के लिए मैं मान लूंगा कि गणनाएँ तात्कालिक हैं) एक अनाम अभिनेता तर्कों की सूची पर एक क्रिया करता है ["A", "B", "C"] -> ["res # A", "res # B", " Res # C "]

 //time:50.003 // // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] // // +---------+ resultList=["res#A","res#B","res#B"] // |anonymous| replyToList=[T-0,T-1,T-2] // +---------+ 

एक पल के बाद, एक अनाम अभिनेता थ्रेड्स के लिए गणना परिणाम वितरित करता है

 //time:50.004 // // T-0 <-----------+ +-------+ generationId=1 // T-1 <---------+ | |Batcher| argList=[] // T-2 <-------+ | | +-------+ replyToList=[] // | | | // | | +---"res#A"--- +---------+ // | +---"res#B"----- |anonymous| // +--"res#C"-------- +---------+ 

एक पल के बाद, सिस्टम अपनी मूल "शुद्ध" स्थिति में वापस आ जाता है

 //time:50.005 // // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] 

बाद में, समय 75 पर, टी -2 स्ट्रीम "डी" की गणना के लिए गुजरता है

 //time:75 // // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=[] // T-2 ----"D"---> +-------+ replyToList=[] 

एक पल के बाद, बैचर जानता है कि "डी" को छोटा करना और टी -2 स्ट्रीम में वापस करना आवश्यक है, इसके अलावा, पीढ़ी के लिए एक टाइमर 1

 // +-----+ timeoutMsg=1 // |Timer| timeout=100 //time:75.001 +-----+ // // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=["D"] // T-2 +-------+ replyToList=[T-2] 

100ms (175ms पर) के बाद, GPars बुनियादी ढांचा प्रतीक्षा अवधि की समाप्ति के टाइमर को सूचित करता है

 // +--"TIMEOUT"-- // | // v // +-----+ timeoutMsg=1 // |Timer| timeout=100 //time:175 +-----+ // // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=["D"] // T-2 +-------+ replyToList=[T-2] 

एक क्षण बाद, टाइमर बैचर को सूचित करता है कि पीढ़ी 1 का समय समाप्त हो गया है और समाप्त () कॉल करके आत्महत्या कर रहा है

 // +-----+ timeoutMsg=1 // +----1-----|Timer| timeout=100 //time:175.001 | +-----+ // v // T-0 +-------+ generationId=1 // T-1 |Batcher| argList=["D"] // T-2 +-------+ replyToList=[T-2] 

एक अनाम अभिनेता बनाया जाता है जो तर्क सूची पर गणना करता है (जिसमें केवल 1 तर्क है)। जनरेशन 1 2 में बदलता है

 //time:175.002 // // T-0 +-------+ generationId=2 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] // // +---------+ argList=["D"] // |anonymous| replyToList=[T-2] // +---------+ 

अभिनेता ने काम किया है

 //time:175.003 // // T-0 +-------+ generationId=2 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] // // +---------+ resultList=["res#D"] // |anonymous| replyToList=[T-2] // +---------+ 

अभिनेता परिणाम देता है

 //time:175.004 // // T-0 +-------+generationId=2 // T-1 |Batcher|argList=[] // T-2 <-------+ +-------+replyToList=[] // | // | +---------+ // +--"res#C"----- |anonymous| // +---------+ 

अपनी मूल "शुद्ध" अवस्था में प्रणाली

 //time:175.005 // // T-0 +-------+ generationId=2 // T-1 |Batcher| argList=[] // T-2 +-------+ replyToList=[] 

समस्या हल करना

बैचप्रोसेसर - "फ़ंक्शन" का इंटरफ़ेस। "बैच मोड" प्रसंस्करण

 import java.util.List; public interface BatchProcessor<ARG, RES> { List<RES> onBatch(List<ARG> argList) throws Exception; } 

बैच - एक वर्ग जो "पैक" तर्क देता है। कोर समाधान

 import groovyx.gpars.actor.*; import groovyx.gpars.actor.impl.MessageStream; import java.util.*; public class Batcher<ARG, RES> extends DynamicDispatchActor { // fixed parameters private final BatchProcessor<ARG, RES> processor; private final int maxBatchSize; private final long batchWaitTimeout; // current state private final List<ARG> argList = new ArrayList<>(); private final List<MessageStream> replyToList = new ArrayList<>(); private long generationId = 0; private Actor lastTimer; public Batcher(BatchProcessor<ARG, RES> processor, int maxBatchSize, long batchWaitTimeout) { this.processor = processor; this.maxBatchSize = maxBatchSize; this.batchWaitTimeout = batchWaitTimeout; } public void onMessage(final ARG elem) { argList.add(elem); replyToList.add(getSender()); if (argList.size() == 1) { lastTimer = new Timer<>(batchWaitTimeout, ++generationId, this).start(); } else if (argList.size() == maxBatchSize) { lastTimer.send("KILL"); lastTimer = null; nextGeneration(); } } public void onMessage(final long timeOutId) { if (generationId == timeOutId) {nextGeneration();} } private void nextGeneration() { new DynamicDispatchActor() { public void onMessage(final Work<ARG, RES> work) throws Exception { List<RES> resultList = work.batcher.onBatch(work.argList); for (int k = 0; k < resultList.size(); k++) { work.replyToList.get(k).send(resultList.get(k)); } terminate(); } }.start().send(new Work<>(processor, new ArrayList<>(argList), new ArrayList<>(replyToList))); argList.clear(); replyToList.clear(); generationId = generationId + 1; } private static class Work<ARG, RES> { public final BatchProcessor<ARG, RES> batcher; public final List<ARG> argList; public final List<MessageStream> replyToList; public Work(BatchProcessor<ARG, RES> batcher, List<ARG> argList, List<MessageStream> replyToList) { this.batcher = batcher; this.argList = argList; this.replyToList = replyToList; } } } 

BatcherDemo बैचर वर्ग का एक प्रदर्शन है। योजनाबद्ध के रूप में भी

 import groovyx.gpars.actor.Actor; import java.io.IOException; import java.util.*; import java.util.concurrent.*; import static java.util.concurrent.Executors.newCachedThreadPool; public class BatcherDemo { public static final int BATCH_SIZE = 3; public static final long BATCH_TIMEOUT = 100; public static void main(String[] args) throws InterruptedException, IOException { final Actor actor = new Batcher<>(new BatchProcessor<String, String>() { public List<String> onBatch(List<String> argList) { System.out.println("onBatch(" + argList + ")"); ArrayList<String> result = new ArrayList<>(argList.size()); for (String arg : argList) { result.add("res#" + arg); } return result; } }, BATCH_SIZE, BATCH_TIMEOUT).start(); ExecutorService exec = newCachedThreadPool(); exec.submit(new Callable<Void>() { // T-0 public Void call() throws Exception { System.out.println(actor.sendAndWait(("A"))); return null; } }); exec.submit(new Callable<Void>() { // T-1 public Void call() throws Exception { Thread.sleep(25); System.out.println(actor.sendAndWait(("B"))); return null; } }); exec.submit(new Callable<Void>() { // T-2 public Void call() throws Exception { Thread.sleep(50); System.out.println(actor.sendAndWait(("C"))); Thread.sleep(25); System.out.println(actor.sendAndWait(("D"))); return null; } }); exec.shutdown(); } } >> onBatch([A, B, C]) >> res#A >> res#B >> res#C >> onBatch([D]) >> res#D 

निष्कर्ष

मेरी राय में, अभिनेता मल्टी-थ्रेडेड प्रिमिटिव प्रोग्रामिंग करने के लिए अच्छे हैं, जो एक जटिल संक्रमण आरेख के साथ परिमित राज्य मशीनें हैं, जो अन्य बातों के अलावा, आने वाली तर्कों पर निर्भर हो सकती हैं।

इस लेख के कुछ उदाहरण विभिन्न स्थानों पर ऑनलाइन पाए गए कोड के रूपांतर हैं, जिनमें gpars.org/guide शामिल हैं ।

दूसरे भाग में हम

• हम प्रस्तावित समाधान की गति को मापते हैं
• एक बड़े RDBMS लेनदेन में व्यक्तिगत लेनदेन से विभिन्न थ्रेड्स के अनुरोधों को जोड़कर JDBC के साथ काम में तेजी लाएं। यही है, हम एक ही कनेक्शन के भीतर बैच नहीं करेंगे, लेकिन विभिन्न कनेक्शनों के बीच।

युपीडी
रोष टिप्पणी के लिए धन्यवाद:
GPars को Java + Groovy मिश्रण में लिखा जाता है।
स्रोत कोड दिखाता है कि ग्रूवी पैकेज लिखे गए हैं
- groovyx.gpars.csp। *
- groovyx.gpars.pa। *
- groovyx.gpars। * (आंशिक रूप से)

संपर्क विवरण

मैं जावा प्रशिक्षण ऑनलाइन करता हूं (यहां प्रोग्रामिंग पाठ्यक्रम हैं ) और जावा कोर कोर्स के रीडिजाइन के भाग के रूप में प्रशिक्षण सामग्री का हिस्सा प्रकाशित करता हूं। आप YouTube चैनल पर दर्शकों के व्याख्यान की वीडियो रिकॉर्डिंग देख सकते हैं, शायद इस लेख में चैनल का वीडियो बेहतर व्यवस्थित है।

स्काइप: गोलोवचक्र्स
ईमेल: GolovachCourses@gmail.com