मैं उन स्थानों पर मौसम डेटा जोड़कर कोड से कुछ पंक्तियों के साथ ट्विटर से संदेशों की एक सतत धारा कैसे प्रिंट कर सकता हूं जहां उनके लेखक रहते हैं? और आप मौसम प्रदाता से अनुरोधों की गति को कैसे सीमित कर सकते हैं ताकि वे हमें ब्लैकलिस्ट न करें?

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

लेख हमारे दिसंबर सम्मेलन डॉटनेक्स्ट 2017 मॉस्को से वागीफ अबिलोव की रिपोर्ट की एक प्रतिलेख पर आधारित है।

मेरा नाम वागीफ है, मैं नॉर्वे की कंपनी माइल्स के लिए काम करता हूं। आज हम अक्का स्ट्रीम्स लाइब्रेरी के बारे में बात करेंगे।

अक्का और रिएक्टिव स्ट्रीम्स काफी संकरे सेटों के प्रतिच्छेदन हैं, और किसी को यह आभास हो सकता है कि यह एक ऐसा आला है, जिसे दर्ज करने के लिए आपको कुछ बेहतरीन ज्ञान रखने की आवश्यकता है, लेकिन सिर्फ विपरीत। और यह लेख यह दिखाने के लिए है कि अक्का धाराओं का उपयोग करके, आप निम्न स्तर की प्रोग्रामिंग से बच सकते हैं जो प्रतिक्रियाशील धाराओं और अक्का.नेट का उपयोग करते समय आवश्यक है। आगे देखते हुए, मैं तुरंत कह सकता हूं: यदि हमारी परियोजना की शुरुआत में, जिस पर हम अक्का का उपयोग करते हैं, हम अक्का धाराओं के अस्तित्व के बारे में जानते थे, हम बहुत अलग तरीके से लिखेंगे, हम समय और कोड दोनों को बचाएंगे।

"शायद सबसे बुरा आप उन लोगों को प्राप्त कर सकते हैं जो आपके एस्पिरिन लेने के लिए दर्द में नहीं हैं।"
मैक्स क्रेमिंस्की
"बंद दरवाजे, सिरदर्द और बौद्धिक जरूरतें"

इससे पहले कि हम तकनीकी विवरणों में शामिल हों, अक्का धाराओं के लिए आपका रास्ता क्या हो सकता है, इसके बारे में थोड़ा सा पता चल सकता है और वहां क्या हो सकता है। एक दिन मैं मैक्स क्रेमिंस्की के ब्लॉग पर आया, जहाँ उन्होंने प्रोग्रामर्स के लिए ऐसा दार्शनिक प्रश्न पूछा: एक प्रोग्रामर के लिए यह बताना असंभव है कि भिक्षुओं क्या हैं। उन्होंने इसे इस तरह समझाया: बहुत बार लोग तुरंत तकनीकी विवरणों पर जाते हैं, यह बताते हुए कि कैसे खूबसूरती से कार्यात्मक प्रोग्रामिंग होती है और मोनाड में कितना समझ है, बिना आश्चर्य के परेशान होने के लिए कि प्रोग्रामर को इसकी आवश्यकता क्यों हो सकती है। एक सादृश्य आरेखित करना, यह पता लगाने की कोशिश किए बिना एस्पिरिन बेचने की कोशिश करना है कि क्या आपके रोगी को दर्द है।

इस सादृश्य का उपयोग करते हुए, मैं निम्नलिखित प्रश्न पूछना चाहता हूं: यदि अक्का धाराएं एस्पिरिन हैं, तो क्या दर्द होना चाहिए जो आपको इसे ले जाएगा?

डेटा स्ट्रीम

सबसे पहले, डेटा धाराओं के बारे में बात करते हैं। प्रवाह काफी सरल, रैखिक हो सकता है।

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

लेकिन प्रवाह अधिक जटिल हो सकता है। यदि आप कई खरगोशों को कंधे से कंधा मिलाकर लगाते हैं, तो हमारे पास पहले से ही प्रवाह का समानांतरीकरण होगा। प्रतिक्रियाशील धाराएँ जो हल करने की कोशिश कर रही हैं, वह ठीक है कि कैसे हम अधिक वैचारिक स्तर पर प्रवाह के साथ संचार कर सकते हैं, अर्थात्, चाहे हम किसी प्रकार के तापमान संवेदक माप के बारे में बात कर रहे हों, जहां रैखिक माप आते हैं। , या हमारे पास रैबिटएमक्यू कतारों के माध्यम से सिस्टम में प्रवेश करने वाले हजारों तापमान सेंसर की निरंतर माप है और सिस्टम लॉग में संग्रहीत है। उपरोक्त सभी को एक संयुक्त धारा माना जा सकता है। यदि आप और भी आगे बढ़ते हैं, तो स्वचालित उत्पादन प्रबंधन (उदाहरण के लिए, कुछ ऑनलाइन स्टोर द्वारा) भी डेटा स्ट्रीम में कम किया जा सकता है, और यह बहुत अच्छा होगा अगर हम इस तरह की स्ट्रीम की योजना बनाने के बारे में बात कर सकते हैं, चाहे वह कितनी भी जटिल हो।

आधुनिक परियोजनाओं के लिए, थ्रेड समर्थन बहुत अच्छा नहीं है। अगर मुझे सही से याद है, तो हारून स्टैनार्ड, जिसका ट्वीट आप तस्वीर में देख रहे हैं, सीएसवी युक्त एक बहु-गीगाबाइट फ़ाइल की एक धारा प्राप्त करना चाहता था, अर्थात्। पाठ, और यह पता चला कि ऐसा कुछ भी नहीं है जिसे आप अतिरिक्त कार्यों के एक समूह के बिना, बस तुरंत ले और उपयोग कर सकते हैं। लेकिन वह बस सीएसवी मूल्यों की एक धारा नहीं पा सका, जिससे वह दुखी हो गया। कुछ समाधान हैं (कुछ विशेष क्षेत्रों के अपवाद के साथ), पुराने तरीकों से बहुत कुछ महसूस किया जाता है, जब हम यह सब खोलते हैं, पढ़ना शुरू करते हैं, बफरिंग करते हैं, तो सबसे खराब स्थिति में, हमें नोटपैड की तरह कुछ मिलता है, जो कहता है कि फ़ाइल बहुत बड़ी है।

एक उच्च वैचारिक स्तर पर, हम सभी डेटा धाराओं को संसाधित करने में लगे हुए हैं, और अक्का धाराएँ आपकी मदद करेंगी यदि:

आप अक्का से परिचित हैं, लेकिन अपने आप को अभिनेता कोड लिखने और उनके समन्वय से जुड़े विवरणों को छोड़ना चाहते हैं;
आप प्रतिक्रियाशील धाराओं से परिचित हैं और उनके विनिर्देश के तैयार-किए गए कार्यान्वयन का उपयोग करना चाहते हैं;
चरणों के लिए अक्का धाराएँ ब्लॉक तत्व आपकी प्रक्रिया को मॉडलिंग करने के लिए उपयुक्त हैं;
आप अपने वर्कफ़्लो के थ्रूपुट चरणों को प्रबंधित करने और गतिशील रूप से परिष्कृत करने के लिए अक्का स्ट्रीम्स बैक प्रेशर (बैकप्रेशर) का लाभ उठाना चाहते हैं।

अभिनेताओं से लेकर अक्का धाराएँ

पहला तरीका एक्टर्स से लेकर अक्का स्ट्रीम्स, मेरा तरीका है।

तस्वीर दिखाती है कि हमने अभिनेता मॉडल का उपयोग क्यों शुरू किया। हम प्रवाह के मैनुअल नियंत्रण से थक गए थे, साझा स्थिति, बस इतना ही। हर कोई जिसने मल्टी-थ्रेडिंग के साथ बड़े सिस्टम के साथ काम किया है, समझता है कि इसमें कितना समय लगता है और इसमें गलती करना कितना आसान है, जो पूरी प्रक्रिया के लिए घातक हो सकता है। यह हमें अभिनेताओं के मॉडल की ओर ले गया। हमें किए गए विकल्प पर पछतावा नहीं है, लेकिन निश्चित रूप से, जब आप काम करना शुरू करते हैं और अधिक प्रोग्रामिंग करते हैं, तो ऐसा नहीं है कि प्रारंभिक उत्साह कुछ और करने का रास्ता देता है, लेकिन आपको यह महसूस करना शुरू हो जाता है कि कुछ और भी प्रभावी ढंग से किया जा सकता है।

“डिफ़ॉल्ट रूप से, उनके संदेशों के प्राप्तकर्ता अभिनेताओं के कोड में दर्ज किए जाते हैं। यदि मैं एक अभिनेता A बनाता हूं जो अभिनेता B को संदेश भेजता है, और आप प्राप्तकर्ता को C के साथ बदलना चाहते हैं, तो सामान्य स्थिति में यह आपके लिए काम नहीं करेगा ”
नोएल वेल्च (अंडरस्कोर .io)

अभिनेताओं ने रचना न करने के लिए आलोचना की। अपने ब्लॉग पर इस बारे में लिखने वाले सबसे पहले नोल्स वेल्च थे, जो अंडरस्कोर के डेवलपर्स में से एक थे। उन्होंने देखा कि अभिनेताओं की प्रणाली कुछ इस तरह दिखती है:

यदि आप किसी भी अतिरिक्त चीजों का उपयोग नहीं करते हैं, जैसे कि निर्भरता इंजेक्शन, इसके प्राप्तकर्ता का पता अभिनेता में सिल दिया जाता है।

जब वे एक-दूसरे को संदेश भेजना शुरू करते हैं, तो यह सब आप पहले से ही निर्धारित करते हैं, प्रोग्रामिंग अभिनेता। और अतिरिक्त चाल के बिना, ऐसी कठोर प्रणाली प्राप्त की जाती है।
अक्का के डेवलपर्स में से एक, रोलैंड कुह्न ने बताया कि आमतौर पर खराब लेआउट का क्या मतलब है। अभिनेता विधि बताई गई विधि पर आधारित है, अर्थात्, अप्रत्यक्ष संदेश: यह शून्य प्रकार की है, अर्थात यह किसी भी चीज़ (या इकाई, भाषा के आधार पर) को वापस नहीं करती है। इसलिए, अभिनेताओं की एक श्रृंखला से प्रक्रिया का विवरण तैयार करना असंभव है। तो आपने बता दिया, फिर क्या? बंद करो। हम शून्य हो गए। आप इसकी तुलना कर सकते हैं, उदाहरण के लिए, LINQ अभिव्यक्तियों के साथ, जहाँ अभिव्यक्ति का प्रत्येक तत्व IQueryable, IEnumerable देता है, और यह सब आसानी से संकलित किया जा सकता है। अभिनेता ऐसा अवसर नहीं देते। इसी समय, रोलांड कुह्न ने इस तथ्य पर आपत्ति जताई कि वे कहते हैं, सिद्धांत में रचना नहीं करते हैं, यह कहते हुए कि वास्तव में वे अन्य तरीकों से संकलित हैं, उसी अर्थ में जिसमें मानव समाज खुद को लेआउट के लिए उधार देता है। यह एक दार्शनिक तर्क की तरह लगता है, लेकिन अगर आप इसके बारे में सोचते हैं, तो सादृश्य समझ में आता है - हाँ, अभिनेता एक-दूसरे को अप्रत्यक्ष संदेश भेजते हैं, लेकिन हम एक-दूसरे के साथ संवाद भी करते हैं, एक-दूसरे के बारे में बात करते हैं, लेकिन साथ ही साथ हम बहुत प्रभावी ढंग से बातचीत करते हैं, यानी हम जटिल सिस्टम बनाते हैं। फिर भी, अभिनेताओं की ऐसी आलोचना मौजूद है।

public class SampleActor : ReceiveActor { public SampleActor() { Idle(); } protected override void PreStart() { /* ... */ } private void Idle() { Receive<Job>(job => /* ... */); } private void Working() { Receive<Cancel>(job => /* ... */); } }

इसके अतिरिक्त, अभिनेता के कार्यान्वयन के लिए कम से कम एक वर्ग लिखने की आवश्यकता होती है यदि आप C # में काम करते हैं, या यदि आप F # में काम करते हैं। ऊपर दिए गए उदाहरण में - बॉयलरप्लेट कोड, जिसे आपको किसी भी मामले में लिखना होगा। हालाँकि यह बहुत बड़ी नहीं है, यह एक निश्चित संख्या की पंक्तियाँ हैं जो आपको हमेशा इस निम्न स्तर पर लिखनी होंगी। यहां मौजूद लगभग सभी कोड एक तरह का समारोह है। क्या होता है जब कोई अभिनेता सीधे संदेश प्राप्त करता है, यहाँ बिल्कुल नहीं दिखाया जाता है। और यह सब लिखने की जरूरत है। यह, ज़ाहिर है, बहुत अधिक नहीं है, लेकिन यह सबूत है कि हम अभिनेताओं के साथ निम्न स्तर पर काम करते हैं, ऐसे शून्य तरीकों का निर्माण करते हैं।

क्या होगा यदि हम एक अलग, उच्च स्तर पर जा सकते हैं, अपने आप से हमारी प्रक्रिया के मॉडलिंग के प्रश्न पूछें, जिसमें विभिन्न स्रोतों से प्रसंस्करण डेटा शामिल हैं जो मिश्रित, परिवर्तित और स्थानांतरित किए गए हैं?

 var results = db.Companies .Join(db.People, c => c.CompanyID, p => p.PersonID, (c, p) => new { c, p }) .Where(z => zcCreated >= fromDate) .OrderByDescending(z => zcCreated) .Select(z => zp) .ToList();

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

 HttpGet pageUrl |> fun s -> Regex.Replace(s, "[^A-Za-z']", " ") |> fun s -> Regex.Split(s, " +") |> Set.ofArray |> Set.filter (fun word -> not (Spellcheck word)) |> Set.iter (fun word -> printfn " %s" word)

(स्रोत)

या, उदाहरण के लिए, कार्यात्मक परिवर्तन। कार्यात्मक प्रोग्रामिंग के बारे में बहुत से लोगों को पसंद है कि आप परिवर्तनों की एक श्रृंखला के माध्यम से डेटा पास कर सकते हैं, और आपको एक स्पष्ट रूप से स्पष्ट कोड प्राप्त होता है, भले ही आप इसे किस भाषा में लिखते हैं। यह पढ़ने में काफी आसान है। चित्र में कोड विशेष रूप से एफ # में लिखा गया है, लेकिन सामान्य तौर पर, शायद, हर कोई समझता है कि यहां क्या हो रहा है।

 val in = Source(1 to 10) val out = Sink.ignore val bcast = builder.add(Broadcast[Int](2)) val merge = builder.add(Merge[Int](2)) val f1,f2,f3,f4 = Flow[Int].map(_ + 10) source ~> f1 ~> bcast ~> f2 ~> merge ~> f3 ~> sink bcast ~> f4 ~> merge ~>

(स्रोत)

फिर यह कैसे? ऊपर दिए गए उदाहरण में, हमारे पास एक स्रोत डेटा स्रोत है, जिसमें 1 से 10 तक पूर्णांक शामिल हैं। यह तथाकथित ग्राफिकल डीएसएल (डोमेन-विशिष्ट भाषा) है। उपरोक्त उदाहरण में डोमेन भाषा के तत्व यूनिडायरेक्शनल एरो सिंबल हैं - ये अतिरिक्त भाषा उपकरण द्वारा परिभाषित ऑपरेटर हैं जो रेखीय रूप से धारा की दिशा दिखाते हैं। हम ट्रांसफॉर्मेशन की एक श्रृंखला के माध्यम से सोर्स पास करते हैं - प्रदर्शन में आसानी के लिए, वे सभी सिर्फ एक संख्या में दस जोड़ते हैं। अगला प्रसारण आता है: हम चैनलों को गुणा करते हैं, अर्थात प्रत्येक संख्या दो चैनलों में प्रवेश करती है। फिर हम फिर से 10 जोड़ते हैं, हमारे डेटा स्ट्रीम को मिलाते हैं, एक नई स्ट्रीम प्राप्त करते हैं, इसमें 10 भी जोड़ते हैं, और यह सब हमारे डेटा स्ट्रीम में जाता है, जिसमें कुछ भी नहीं होता है। यह वास्तविक कोड है जो इस भाषा में लागू किया गया, अकाला धाराओं का हिस्सा स्काला में लिखा गया है। यही है, आप अपने डेटा के परिवर्तन के चरणों को निर्दिष्ट करते हैं, इंगित करते हैं कि उनके साथ क्या करना है, स्रोत, स्टॉक, कुछ चौकियों को निर्दिष्ट करें और फिर ग्राफिक डीएसएल का उपयोग करके ऐसा ग्राफ़ बनाएं। यह एक कार्यक्रम के लिए सभी कोड है। कोड की कुछ पंक्तियों से पता चलता है कि प्रक्रिया में क्या हो रहा है।

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

 var runnable = Source .From(Enumerable.Range(1, 1000)) .Via(Flow.Create<int>().Select(x => x * 2) .To(Sink.ForEach<int>(x => Console.Write(x.ToString));

उपरोक्त उदाहरण दिखाता है कि यह C # में कैसे दिख सकता है। सबसे सरल तरीका: हमारे पास एक डेटा स्रोत है - ये 1 से 1000 तक संख्याएं हैं (जैसा कि आप देख सकते हैं, अक्का धाराओं में कोई भी IEnumerable डेटा प्रवाह का स्रोत बन सकता है, जो बहुत सुविधाजनक है)। हम कुछ सरल गणना करते हैं, कहते हैं, हम दो से गुणा करते हैं, और फिर डेटा स्ट्रीम पर यह सब स्क्रीन पर प्रदर्शित होता है।

 var graph = GraphDsl.Create(builder => { var bcast = builder.Add(new Broadcast<int>(2)); var merge = builder.Add(new Merge<int, int>(2)); var count = Flow.FromFunction(new Func<int, int>(x => 1)); var sum = Flow.Create<int>().Sum((x, y) => x + y); builder.From(bcast.Out(0)).To(merge.In(0)); builder.From(bcast.Out(1)).Via(count).Via(sum).To(merge.In(1)); return new FlowShape<int, int>(bcast.In, merge.Out); });

ऊपर दिए गए उदाहरण में जो दिखाया गया है उसे "C # में ग्राफिकल डीएसएल" कहा जाता है। वास्तव में, यहां कोई ग्राफिक्स नहीं है, यह स्काला के साथ एक बंदरगाह है, लेकिन सी # में इस तरह से ऑपरेटरों को परिभाषित करने का कोई तरीका नहीं है, इसलिए यह थोड़ा अधिक बोझिल दिखता है, लेकिन फिर भी यह समझने के लिए पर्याप्त कॉम्पैक्ट है कि यहां क्या हो रहा है। तो, हम एक निश्चित ग्राफ़ बना रहे हैं (विभिन्न प्रकार के ग्राफ़ हैं, यहाँ इसे अलग-अलग घटकों से फ़्लोशैप कहा जाता है), जहाँ एक डेटा स्रोत होता है और कुछ परिवर्तन होते हैं। हम एक चैनल में डेटा भेजते हैं जिसमें हम गणना करते हैं, अर्थात, प्रसारित किए जाने वाले डेटा तत्वों की संख्या, और दूसरे में हम योग उत्पन्न करते हैं और फिर हम इसे मिलाते हैं। आगे हम पूर्णांक को संसाधित करने की तुलना में अधिक दिलचस्प उदाहरण देखेंगे।
यह पहला रास्ता है जो आपको अक्का धाराओं तक ले जा सकता है, यदि आपके पास एक अभिनेता मॉडल के साथ काम करने का अनुभव है और आपने सोचा है कि क्या मैन्युअल रूप से लिखना है, यहां तक कि सबसे सरल अभिनेता भी। दूसरा तरीका जो अक्का स्ट्रीम्स का है, वह रिएक्टिव स्ट्रीम्स के माध्यम से आता है।

प्रतिक्रियाशील धाराओं से अक्का धाराओं तक

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

प्रतिक्रियाशील धाराएं आपको अनुक्रम में संभावित असीमित संख्या में तत्वों को संसाधित करने और गैर-अवरुद्ध दबाव वाले घटकों के बीच अतुल्यकालिक रूप से तत्वों को स्थानांतरित करने की अनुमति देती हैं।

प्रतिक्रियाशील धाराओं के निर्माण के सर्जक की सूची काफी प्रभावशाली है: यहां नेटफ्लिक्स और ओरेकल और ट्विटर हैं।

विभिन्न भाषाओं और प्लेटफार्मों में कार्यान्वयन को यथासंभव सरल बनाने के लिए विनिर्देश बहुत सरल है। प्रतिक्रियाशील धाराओं एपीआई के मुख्य घटक:

प्रकाशक
ग्राहक
अंशदान
प्रोसेसर

अनिवार्य रूप से, इस विनिर्देश का अर्थ यह नहीं है कि आप स्वयं इन इंटरफेस को लागू करना शुरू करेंगे। यह समझा जाता है कि कुछ पुस्तकालय डेवलपर्स हैं जो आपके लिए ऐसा करेंगे। और अक्का स्ट्रीम इस विनिर्देश के कार्यान्वयन में से एक है।

 public interface IPublisher<out T> { void Subscribe(ISubscriber<T> subscriber); } public interface ISubscriber<in T> { void OnSubscribe(ISubscription subscription); void OnNext(T element); void OnError(Exception cause); void OnComplete(); }

इंटरफेस, जैसा कि आप उदाहरण से देख सकते हैं, वास्तव में बहुत सरल हैं: उदाहरण के लिए, प्रकाशक में केवल एक विधि है - "सदस्यता"। सबस्क्राइबर, सब्सक्राइबर, इस घटना के लिए कुछ प्रतिक्रियाएँ शामिल करता है।

 public interface ISubscription { void Request(long n); void Cancel(); } public interface IProcessor<in T1, out T2> : ISubscriber<T1>, IPublisher<T2> { }

अंत में, सदस्यता में दो विधियाँ होती हैं - "प्रारंभ" और "मना"। प्रोसेसर किसी भी नए तरीके को परिभाषित नहीं करता है; यह एक प्रकाशक और एक ग्राहक को जोड़ती है।

अन्य धारा के कार्यान्वयन के अलावा प्रतिक्रियाशील धाराएं क्या निर्धारित करती हैं? रिएक्टिव स्ट्रीम्स पुश और मॉडल को जोड़ती हैं। समर्थन के लिए, यह सबसे कुशल प्रदर्शन परिदृश्य है। मान लीजिए आपके पास एक धीमा डेटा ग्राहक है। इस मामले में, उसके लिए धक्का घातक हो सकता है: यदि आप उसे भारी मात्रा में डेटा भेजते हैं, तो वह उन्हें संसाधित करने में सक्षम नहीं होगा। पुल का उपयोग करना बेहतर है ताकि ग्राहक स्वयं प्रकाशक से डेटा खींच ले। लेकिन अगर प्रकाशक धीमा है, तो यह पता चलता है कि ग्राहक हर समय अवरुद्ध रहता है, हर समय प्रतीक्षा करता है। एक मध्यवर्ती समाधान कॉन्फ़िगरेशन हो सकता है: हमारे पास एक कॉन्फ़िगरेशन-फ़ाइल है जिसमें हम निर्धारित करते हैं कि उनमें से कौन सा तेज है। और अगर उनकी गति बदल जाती है?

तो, सबसे सुरुचिपूर्ण कार्यान्वयन वह है जिसमें हम गतिशील रूप से पुश और पुल मॉडल बदल सकते हैं।

(स्रोत (Apache Flink))

आरेख दिखाता है कि यह कैसे हो सकता है। यह डेमो Apache Flink का उपयोग करता है। येलो एक प्रकाशक, डेटा निर्माता हैं, उन्हें उनकी क्षमता का लगभग 50% सेट किया गया था। ग्राहक सबसे अच्छी रणनीति चुनने की कोशिश करता है - यह धक्का लगता है। फिर हम सब्सक्राइबर को लगभग 20% की गति पर रीसेट करते हैं, और वह खींचने के लिए स्विच करता है। फिर हम 100% जाते हैं, फिर से 20% वापस खींचते हैं, पुल मॉडल आदि। यह सब डायनेमिक्स में होता है, आपको सेवा को रोकने की ज़रूरत नहीं है, कॉन्फ़िगरेशन में कुछ दर्ज करें। यह एक उदाहरण है कि अक्का धाराओं में दबाव कितना काम करता है।

अक्का धाराएँ सिद्धांत

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

डेटा का स्रोत (स्रोत) - एक आउटपुट के साथ प्रसंस्करण का चरण।
सिंक एक सिंगल-एंट्री प्रोसेसिंग स्टेप है।
चेकपॉइंट (प्रवाह) - एक इनपुट और एक आउटपुट के साथ प्रसंस्करण चरण। कार्यात्मक परिवर्तन यहां होते हैं, और जरूरी नहीं कि स्मृति में: यह हो सकता है, उदाहरण के लिए, वेब सेवा के लिए कॉल, समानता के कुछ तत्वों के लिए, बहु-थ्रेडेड।

इन तीन प्रकारों में से, ग्राफ़ बनाए जा सकते हैं। ये पहले से ही अधिक जटिल प्रसंस्करण चरण हैं, जो स्रोतों, नालियों और चौकियों से निर्मित हैं। लेकिन हर ग्राफ को निष्पादित नहीं किया जा सकता है: यदि इसमें छेद हैं, अर्थात इनपुट और आउटपुट खुले हैं, तो यह ग्राफ़ नहीं चलता है।
एक ग्राफ एक रननेबल ग्राफ है, अगर यह बंद है, अर्थात, प्रत्येक इनपुट के लिए एक आउटपुट है: यदि डेटा दर्ज किया गया है, तो यह कहीं चला गया होगा।

अक्का धाराओं में अंतर्निहित स्रोत हैं: तस्वीर में आप देखते हैं कि उनमें से कितने हैं। उनके नाम एक-से-एक हैं और प्रतिबिंबित करते हैं कि स्काला या जेवीएम में कुछ .NET-विशिष्ट उपयोगी स्रोतों के अपवाद हैं। पहले दो (FromEnumerator और From) कुछ सबसे महत्वपूर्ण हैं: किसी भी नंबरिंग, किसी भी ienumerable को एक धारा स्रोत में बदल दिया जा सकता है।

अंतर्निहित नालियाँ हैं: उनमें से कुछ LINQ विधियों से मिलते जुलते हैं, उदाहरण के लिए, First, Last, FirstOrDefault। बेशक, जो कुछ भी आपको मिलता है, आप फाइलों में, धाराओं में, अक्का धाराओं में नहीं, बल्कि .NET स्ट्रीम में डंप कर सकते हैं। और फिर, यदि आपके पास आपके सिस्टम में कोई भी अभिनेता है, तो आप उन्हें इनपुट और सिस्टम के आउटपुट पर दोनों का उपयोग कर सकते हैं, अर्थात, यदि आप चाहें, तो अपने तैयार सिस्टम में इसे एम्बेड कर सकते हैं।

और बड़ी संख्या में अंतर्निर्मित चौकियाँ हैं, जो, शायद, LINQ की और भी अधिक याद दिलाती हैं, क्योंकि यहाँ Select, और SelectMany, और GroupBy हैं, अर्थात, जिनका उपयोग हम LINQ में काम करने के लिए करते हैं।

उदाहरण के लिए, Select in Scala को SelectAsync कहा जाता है: यह पर्याप्त शक्तिशाली है क्योंकि यह समानता के स्तर को एक तर्क के रूप में लेता है। यही है, आप यह संकेत दे सकते हैं कि, उदाहरण के लिए, चयन दस थ्रेड्स के समानांतर में कुछ वेब सेवा को डेटा भेजता है, फिर वे सभी एकत्र किए जाते हैं और पारित होते हैं। वास्तव में, आप कोड की एक पंक्ति के साथ चौकी के स्केलिंग की डिग्री निर्धारित करते हैं।

एक प्रवाह घोषणा इसकी निष्पादन योजना है, अर्थात्, एक ग्राफ, यहां तक कि एक रन भी, बस उसी तरह निष्पादित नहीं किया जा सकता है - इसे भौतिक बनाने की आवश्यकता है। एक त्वरित प्रणाली होनी चाहिए, एक अभिनेता प्रणाली, आपको इसे एक धारा देना होगा, निष्पादन के लिए यह योजना, और फिर इसे निष्पादित किया जाएगा। इसके अलावा, रन टाइम पर यह अत्यधिक ऑप्टिमाइज़ किया जाता है, जब आप किसी डेटाबेस में LINQ एक्सप्रेशन भेजते हैं: एक प्रदाता आपके SQL को अधिक कुशल डेटा आउटपुट के लिए ऑप्टिमाइज़ कर सकता है, अनिवार्य रूप से क्वेरी कमांड को दूसरे के साथ बदल सकता है। अक्का धाराओं के साथ भी: संस्करण 2.0 से शुरू होकर आप कुछ निश्चित चौकियों को निर्धारित कर सकते हैं, और सिस्टम समझ जाएगा कि उनमें से कुछ को संयुक्त किया जा सकता है ताकि वे एक अभिनेता (ऑपरेटर संलयन) द्वारा निष्पादित हो सकें। चेकपॉइंट्स, एक नियम के रूप में, प्रसंस्करण तत्वों के क्रम को बनाए रखते हैं।

 var results = db.Companies .Join(db.People, c => c.CompanyID, p => p.PersonID, (c, p) => new { c, p }) .Where(z => zcCreated >= fromDate) .OrderByDescending(z => zcCreated) .Select(z => zp) .ToList();

ऊपर के उदाहरण में LINQ अभिव्यक्ति के अंतिम टॉलिस्ट एलिमेंट की तुलना स्ट्रीम स्ट्रीम मटीरियलाइजेशन से की जा सकती है। अगर हम ToList नहीं लिखते हैं, तो हमें एक अनमैमेटाइज्ड LINQ एक्सप्रेशन मिलता है, जिससे डेटा SQL सर्वर या Oracle में ट्रांसफर नहीं होगा, क्योंकि ज्यादातर LINQ प्रोवाइडर्स तथाकथित डिफरेंट क्वेरी एक्जीक्यूशन (देरी से क्वेरी एक्जीक्यूशन), t को सपोर्ट करते हैं। यानी, अनुरोध को केवल तभी निष्पादित किया जाता है जब कुछ परिणाम देने के लिए एक आदेश दिया जाता है। अनुरोध के आधार पर - एक सूची या पहला परिणाम - सबसे प्रभावी टीम बनाई जाएगी। जब हम ToList कहते हैं, तो हम LINQ प्रदाता से हमें तैयार परिणाम देने के लिए कहते हैं।

 var runnable = Source .From(Enumerable.Range(1, 1000)) .Via(Flow.Create<int>().Select(x => x * 2) .To(Sink.ForEach<int>(x => Console.Write(x.ToString));

अक्का धाराएँ उसी तरह काम करती हैं। तस्वीर में हमारा लॉन्च किया गया ग्राफ है, जिसमें चौकियों और अपवाह का स्रोत है, और हम अब इसे चलाना चाहते हैं।

 var runnable = Source .From(Enumerable.Range(1, 1000)) .Via(Flow.Create<int>().Select(x => x * 2) .To(Sink.ForEach<int>(x => Console.Write(x.ToString)); var system = ActorSystem.Create("MyActorSystem"); using (var materializer = ActorMaterializer.Create(system)) { await runnable.Run(materializer); }

ऐसा होने के लिए, हमें अभिनेताओं की एक प्रणाली बनाने की आवश्यकता है, इसमें एक भौतिकवादी है, हमारे ग्राफ को उसके पास करें, और वह इसे निष्पादित करेगा। यदि हम इसे दोबारा बनाते हैं, तो यह इसे फिर से निष्पादित करेगा, और अन्य परिणाम प्राप्त हो सकते हैं।

प्रवाह के भौतिककरण के अलावा, अक्का धाराओं के भौतिक भाग के बारे में बात करते हुए, यह भौतिक मूल्यों का उल्लेख करने योग्य है।

 var output = new List<int>(); var source1 = Source.From(Enumerable.Range(1, 1000)); var sink1 = Sink.ForEach<int>(output.Add); IRunnableGraph<NotUsed> runnable1 = source1.To(sink1); var source2 = Source.From(Enumerable.Range(1, 1000)); var sink2 = Sink.Sum<int>((x,y) => x + y); IRunnableGraph<Task<int>> runnable2 = source2.ToMaterialized(sink2, Keep.Right);

जब हमारे पास एक ऐसी धारा होती है जो चौकियों से होकर नाले तक जाती है, तो अगर हम किसी भी मध्यवर्ती मूल्यों का अनुरोध नहीं करते हैं, तो वे हमारे लिए उपलब्ध नहीं हैं, क्योंकि इसे सबसे कुशल तरीके से निष्पादित किया जाएगा। यह एक ब्लैक बॉक्स की तरह है। लेकिन कुछ मध्यवर्ती मूल्यों को बाहर निकालना हमारे लिए दिलचस्प हो सकता है, क्योंकि बाईं ओर प्रत्येक बिंदु पर कुछ मूल्य आते हैं, दूसरे मूल्य दाईं ओर निकलते हैं, और आप जो रुचि रखते हैं उसे इंगित करने के लिए एक ग्राफ़ निर्दिष्ट कर सकते हैं। ऊपर दिए गए उदाहरण में, एक रन-इन ग्राफ जिसमें NotUsed का संकेत दिया गया है, अर्थात कोई भी भौतिक मूल्य हमारे हित में नहीं है। नीचे हम इसे संकेत के साथ बनाते हैं कि अपवाह के दाईं ओर, अर्थात् सभी परिवर्तनों को पूरा करने के बाद, हमें भौतिक मूल्यों को बाहर करने की आवश्यकता है। हमें ग्राफ मिलता है टास्क - एक कार्य, जिसके पूरा होने पर हमें एक इंट मिलता है, जो कि इस ग्राफ के अंत में होता है। आप प्रत्येक पैराग्राफ में इंगित कर सकते हैं कि आपको किसी प्रकार के भौतिक मूल्यों की आवश्यकता है, यह सब धीरे-धीरे एकत्र किया जाएगा।

डेटा को अक्का धाराओं में स्थानांतरित करने के लिए या उन्हें वहां से बाहर निकालने के लिए, निश्चित रूप से, बाहरी दुनिया के साथ किसी प्रकार की बातचीत की आवश्यकता होती है। एंबेडेड स्रोत चरणों में प्रतिक्रियाशील डेटा धाराओं की एक विस्तृत श्रृंखला होती है:

Source.FromEnumerator और Source.From आपको IEnumerable लागू करने वाले किसी भी स्रोत से डेटा स्थानांतरित करने की अनुमति देता है;
अनफोल्डेड और अनफोल्डेन्क्स फ़ंक्शन फ़ंक्शन गणना के परिणाम उत्पन्न करते हैं बशर्ते कि यह गैर-शून्य मान लौटाता है;
FromInputStream एक स्ट्रीम को रूपांतरित करता है;
FromFile प्रतिक्रियाशील धारा में फ़ाइल की सामग्री को पार्स करता है;
ActorPublisher अभिनेता संदेशों को धर्मान्तरित करता है।

जैसा कि मैंने पहले ही कहा था। .NET डेवलपर्स के लिए यह एन्यूमरेटर या IEnumerable का उपयोग करने के लिए बहुत उत्पादक है, लेकिन कभी-कभी यह डेटा का उपयोग करने के लिए बहुत आदिम, अक्षम तरीका भी है। अधिक जटिल स्रोतों में बड़ी मात्रा में डेटा को विशेष कनेक्टर्स की आवश्यकता होती है। ऐसे कनेक्टर लिखे जाते हैं। एक ओपन सोर्स प्रोजेक्ट अल्पाका है, जो मूल रूप से स्काला में दिखाई दिया और अब .NET में है। इसके अलावा, अक्का के पास लगातार कलाकार हैं, और उनकी अपनी धाराएं हैं जिनका उपयोग किया जा सकता है (उदाहरण के लिए, अक्का पर्सिस्टेंस क्वेरी अक्का इवेंट जर्नल का कंटेंट फ्लो बनाती है)।

यदि आप स्काला के साथ काम करते हैं, तो सबसे आसान तरीका आपके लिए है: बड़ी संख्या में कनेक्टर हैं, और आप निश्चित रूप से अपने स्वाद के लिए कुछ पाएंगे। जानकारी के लिए, काफ्का तथाकथित प्रतिक्रियाशील काफ्का है, काफ्का धाराएँ नहीं। काफ्का स्ट्रीम, जहाँ तक मुझे पता है, पीठ के दबाव का समर्थन नहीं करता है। प्रतिक्रियाशील काफ्का कफ़्का से एक धारा कार्यान्वयन है जो प्रतिक्रियाशील धाराओं का समर्थन करता है।

अल्पाका .NET कनेक्टर की सूची अधिक विनम्र है, लेकिन इसे फिर से भरना है, और प्रतियोगिता का एक तत्व है। माइक्रोसॉफ्ट से डेविड फाउलर का एक आधे साल पुराना ट्वीट है, जिन्होंने कहा कि सिग्नलआर अब रिएक्टिव एक्सटेंशन्स के साथ डेटा का आदान-प्रदान कर सकता है, और अक्का डेवलपर्स में से एक ने जवाब दिया कि यह वास्तव में कुछ समय के लिए अक्का धाराओं में था। अक्का Microsoft Azure की विभिन्न सेवाओं का समर्थन करता है। CSV हारून स्टैनार्ड की हताशा का परिणाम है जब उन्होंने पाया कि CSV के लिए कोई अच्छी धारा नहीं है: अब CSV XML के लिए अक्का की अपनी एक धारा है। AMQP है (वास्तव में, RabbitMQ), यह विकास के अधीन है, लेकिन उपयोग के लिए उपलब्ध है, यह काम करता है। काफ्का भी विकास के अधीन है। इस सूची का विस्तार जारी रहेगा।

विकल्पों के बारे में कुछ शब्द, चूंकि आप डेटा धाराओं के साथ काम करते हैं, अक्का धाराएं, इन धाराओं को संभालने का एकमात्र तरीका नहीं है। सबसे अधिक संभावना है, आपकी परियोजना में, धागे को लागू करने का विकल्प कई अन्य कारकों पर निर्भर करेगा जो कुंजी बन सकते हैं। उदाहरण के लिए, यदि आप Microsoft Azure और Orleans के साथ बहुत काम करते हैं, तो आपको आभासी अभिनेताओं के समर्थन के साथ अपनी परियोजना की जरूरतों में व्यवस्थित किया जाता है, या, जैसा कि वे उन्हें अनाज कहते हैं, तो उनका अपना कार्यान्वयन है जो प्रतिक्रियाशील धाराओं विनिर्देश - ऑरलियन्स धाराओं को पूरा नहीं करता है, जो यह आपके लिए निकटतम होगा, और यह आपके लिए इस पर ध्यान देने के लिए समझ में आता है। यदि आप टीपीएल के साथ बहुत काम करते हैं, तो टीपीएल डेटाफ्लो है - यह अक्का धाराओं के सबसे निकटतम सादृश्य हो सकता है: डेटा धाराओं को जोड़ने के लिए प्राइमरी भी हैं, साथ ही बफरिंग और बैंडविड्थ लिमिटिंग टूल (बाउंडेडसिटी, मैक्समेसजपेरटैस्क)। यदि अभिनेता मॉडल के विचार आपके करीब हैं, तो अक्का धाराएं इसे संबोधित करने और प्रत्येक अभिनेता को मैन्युअल रूप से लिखने के बिना समय की एक महत्वपूर्ण राशि को बचाने का एक तरीका है।

कार्यान्वयन उदाहरण: इवेंट लॉग स्ट्रीम

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

हम विभिन्न मीडिया फ़ाइलों को क्लाउड पर अपलोड करते हैं। यह परियोजना का प्रारंभिक चरण था: यहां अंतिम 15 मिनट 23 फाइलों में, जिनमें से 7 त्रुटियां हैं। अब व्यावहारिक रूप से कोई त्रुटि नहीं है और फ़ाइलों की संख्या बहुत बड़ी है - सैकड़ों हर कुछ मिनटों के माध्यम से जाते हैं। यह सब किबाना डैशबोर्ड में निहित है।
किबाना एलियस्टिक्स खोज से डेटा पढ़ता है, और चूंकि प्राथमिक डेटा एलिस्टिक्स खोज में संग्रहीत किया जाता है, इसलिए इस अनुक्रमणिका के कार्यान्वयन के लिए आवश्यक है कि आप इसे हटा सकते हैं और इसे फिर से भरने के लिए एक आदेश जारी कर सकते हैं। चूंकि परियोजना विकास के अधीन है, यह हमें डेटा प्रारूपों को बदलने, उन्हें नए मूल्यों के साथ विस्तार करने की अनुमति देता है, अर्थात सूचकांक को लगातार अद्यतन करने की आवश्यकता होती है। यह इवेंट जर्नल अक्का की सामग्री के साथ फिर से भर दिया गया है, जो Microsoft SQL सर्वर डेटाबेस में संग्रहीत है। पहले से सहेजे गए ईवेंट और रीयल-टाइम इवेंट दोनों को वर्तमान ऑपरेशन पैनल में प्रदर्शित किया जाना चाहिए।

 CREATE TABLE EventJournal ( Ordering BIGINT IDENTITY(1,1) PRIMARY KEY NOT NULL, PersistenceID NVARCHAR(255) NOT NULL, SequenceNr BIGINT NOT NULL, Timestamp BIGINT NOT NULL, IsDeleted BIT NOT NULL, Manifest NVARCHAR(500) NOT NULL, Payload VARBINARY(MAX) NOT NULL, Tags NVARCHAR(100) NULL CONSTRAINT QU_EventJournal UNIQUE (PersistenceID, SequenceNr) )

इसे प्राप्त करने के लिए, हमें एक ओर, SQL सर्वर से लिए गए डेटा को फिर से लिखने की जरूरत है, जिसमें कुछ इवेंटस्टोर में निरंतर कलाकार अक्का, EventJournal शामिल हैं। चित्र एक विशिष्ट घटना-स्थल को दर्शाता है।

और दूसरी तरफ, डेटा वास्तविक समय में आता है। और यह पता चला है कि एक सूचकांक लिखने के लिए, हमें डेटाबेस से डेटा पढ़ने की आवश्यकता है, साथ ही वास्तविक समय डेटा आता है, और कुछ बिंदु पर हमें समझने की आवश्यकता है: यहां से डेटा समाप्त हो गया, यह नया है। इस सीमावर्ती क्षण के लिए अतिरिक्त सत्यापन की आवश्यकता होती है ताकि कुछ भी न खोएं और दो बार कुछ भी रिकॉर्ड न करें। यही है, यह किसी तरह से बल्कि जटिल निकला। मेरे सहयोगी और मैं खुश नहीं थे कि क्या हो रहा है। यह बहुत ही जटिल कोड नहीं है, बल्कि बहुत ही डरावना है। जब तक हमें याद नहीं आया कि अक्का में लगातार कलाकार जिद का समर्थन करते हैं।

यह सिर्फ उन्हें स्रोत से ऊपर डेटा स्ट्रीम के रूप में प्राप्त करने का अवसर है, वे डेटाबेस से आते हैं या वास्तविक समय में प्राप्त होते हैं।

अंतर्निहित क्वेरी (दृढ़ता प्रश्न):

AllPersistencelds
CurrentPersistencelds
EventsByPersistenceld
CurrentEventsByPersistenceld
EventsByTag
CurrentEventsByTag

और ऐसे कई तरीके हैं, जिनका हम उपयोग कर सकते हैं, उदाहरण के लिए, वर्तमान विधि है - यह एक स्नैपशॉट है, समय के साथ ऐतिहासिक रूप से एक निश्चित बिंदु तक डेटा। और इस उपसर्ग के बिना, पहले और वास्तविक लोगों सहित। हमें EventByTag की जरूरत थी।

 let system = mailbox.Context.System let queries = PersistenceQuery.Get(system) .ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier) let mat = ActorMaterializer.Create(system) let offset = getCurrentOffset client config let ks = KillSwitches.Shared "persistence-elastic" let task = queries.EventsByTag(PersistenceUtils.anyEventTag, offset) .Select(fun e -> ElasticTypes.EventEnvelope.FromAkka e) .GroupedWithin(config.BatchSize, config.BatchTimeout) .Via(ks.Flow()) .RunForeach((fun batch -> processItems client batch), mat) .ContinueWith(handleStreamError mailbox, TaskContinuationOptions.OnlyOnFaulted) |> Async.AwaitTaskVoid

और यह पता चला कि हमारे पास पर्याप्त कोड था। यह एफ # में लिखा गया था, लेकिन सी # में यह उसी कॉम्पैक्ट के बारे में था। हमें EventByTag मिलता है, अक्का स्ट्रीम बिल्ट-इन ब्लॉक का उपयोग करते हैं, और इन सभी से हमें वह डेटा मिलता है, जिसे हम एलिस्टिक्स में हेलमेट करते हैं। यही है, हमने किसी और के डेटा स्ट्रीम के कार्यान्वयन का लाभ उठाया, और इसने हमें यह भूलने की अनुमति दी कि हमारा डेटा कहाँ है, यह कहाँ से आता है - डेटाबेस से या यह वास्तविक समय में होता है। इस कार्यान्वयन ने हमें एक ही अनुरोध के साथ यह सब दिया।

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

कार्यान्वयन उदाहरण: जेट ट्वीट्स

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

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

यह सब मेरे GitHub खाते में, AkkaStreamsDemo में है । खोलें और देखें (या आप अभी से रिपोर्ट रिकॉर्डिंग देखना शुरू कर सकते हैं )।

चलो एक साधारण से शुरू करते हैं। सबसे पहले, मैं Twitter से सीधे डेटा पढ़ना चाहता हूं: Program.cs फ़ाइल में

 var useCachedTweets = false

यदि मुझे ट्विटर से प्रतिबंधित कर दिया जाता है, तो मैंने ट्वीट को कैश कर दिया है, वे तेज हैं। शुरू करने के लिए, हम कुछ RunnableGraph बनाते हैं।

 public static IRunnableGraph<IActorRef> CreateRunnableGraph() { var tweetSource = Source.ActorRef<ITweet>(100, OverflowStrategy.DropHead); var formatFlow = Flow.Create<ITweet>().Select(Utils.FormatTweet); var writeSink = Sink.ForEach<string>(Console.WriteLine); return tweetSource.Via(formatFlow).To(writeSink); }

( स्रोत )

हमारे यहां ट्वीट्स का एक स्रोत है, जो एक अभिनेता से आता है। मैं आपको दिखाता हूं कि हम इन ट्वीट्स को वहां कैसे खींचते हैं, उन्हें प्रारूपित करें (ट्वीट प्रारूप केवल लेखक को एक ट्वीट देता है) और फिर इसे स्क्रीन पर लिखें।

StartTweetStream - यहाँ हम Tweetinvi लाइब्रेरी का उपयोग करेंगे।

 public static void StartTweetStream(IActorRef actor) { var stream = Stream.CreateSampleStream(); stream.TweetReceived += (_, arg) => { arg.Tweet.Text = arg.Tweet.Text.Replace("\r", " ").Replace("\n", " "); var json = JsonConvert.SerializeObject(arg.Tweet); File.AppendAllText("tweets.txt", $"{json}\r\n"); actor.Tell(arg.Tweet); }; stream.StartStream(); }

( स्रोत )

क्रिएटसम्प्लीस्ट्रीम के माध्यम से हमें सैंपल ट्वीट मिलते हैं, वे बहुत उच्च गति पर जारी नहीं किए जाते हैं। इस सब से, हम चुनते हैं कि हमें क्या चाहिए और एक अभिनेता का निर्माण करें जो कहता है: "इस ट्वीट को स्वीकार करें।" अगला हमें IEnumerable प्राप्त करने की आवश्यकता है, ताकि अंत में हमें स्रोत मिल सके।

और TweetEnumerator बहुत सरल दिखता है: हमारे पास ट्वीट्स का एक संग्रह है, और हमें वर्तमान, MoveNext, रीसेट, और अच्छे नागरिक होने का निपटान करने की आवश्यकता है। यदि हम इसे चलाते हैं, तो हम एक वास्तविक समय का उदाहरण देखेंगे। गैर-मुद्रण का एक बहुत कुछ है, क्योंकि यह विभिन्न गैर-लैटिन देशों से है। यह हमारे कार्यक्रम का सबसे आसान संस्करण है।

अब हम useCachedTweets के मान को सही में बदलते हैं, और यहाँ जटिलताएँ शुरू होती हैं। कैशेड ट्वीट्स एक ही बात है, केवल मेरे पास 50,000 ट्वीट्स की एक फाइल है जिसे मैंने पहले ही चुना है, सहेजा है, हम उनका उपयोग करेंगे। मैंने उन ट्वीट्स को चुनने की कोशिश की, जिनमें उनके लेखकों के भौगोलिक निर्देशांक पर डेटा है, जिनकी हमें आवश्यकता होगी। अगला कदम यह है कि हम ट्वीट को समानांतर करना चाहते हैं। निष्पादन के बाद, हम पहले सूची में ट्वीट के मालिक होंगे, और फिर निर्देशांक।

TweetsWithBroadcast:

 var graph = GraphDsl.Create(b => { var broadcast = b.Add(new Broadcast<ITweet>(2)); var merge = b.Add(new Merge<string>(2)); b.From(broadcast.Out(0)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.CreatedBy)) .Via(formatUser) .To(merge.In(0)); b.From(broadcast.Out(1)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.Coordinates)) .Via(formatCoordinates) .To(merge.In(1)); return new FlowShape<ITweet, string>(broadcast.In, merge.Out); });

( स्रोत )

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

हमारे डेमो में अगला कदम चीजों को थोड़ा जटिल करना है। चलिए बैंडविड्थ को बदलना शुरू करते हैं।

 var graph = GraphDsl.Create(b => { var broadcast = b.Add(new Broadcast<ITweet>(2)); var merge = b.Add(new Merge<string>(2)); b.From(broadcast.Out(0)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.CreatedBy) .Throttle(10, TimeSpan.FromSeconds(1), 1, ThrottleMode.Shaping)) .Via(formatUser) .To(merge.In(0)); b.From(broadcast.Out(1)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.Coordinates) .Buffer(10, OverflowStrategy.DropNew) .Throttle(1, TimeSpan.FromSeconds(1), 10, ThrottleMode.Shaping)) .Via(formatCoordinates) .To(merge.In(1)); return new FlowShape<ITweet, string>(broadcast.In, merge.Out); });}

( स्रोत )

पहले चैनल में, हमारे पास प्रति सेकंड 10 ट्वीट्स तक की बैंडविड्थ सीमा होती है, दूसरे चैनल पर हमारे पास 10 सेकंड के बफर के साथ प्रति सेकंड एक ट्वीट होता है। जब हम बफर को निर्दिष्ट करते हैं, तो हमें इस बात की रणनीति का संकेत देना चाहिए कि क्या बफर भरा हुआ है। यह, वैसे, सामान्य रूप से अक्का धाराओं और प्रतिक्रियाशील धाराओं को अलग करता है: इस रणनीति का विकल्प एक होना चाहिए। कई मामलों में, जब हम थ्रेड्स के साथ काम करते हैं, तो हमें नहीं पता होता है कि अगर हमारे साथ कुछ ओवरफ्लो होने लगे तो क्या होगा। यहां हम चुन सकते हैं, उदाहरण के लिए, यदि यह महत्वपूर्ण डेटा है, तो संपूर्ण स्ट्रीम त्रुटि संदेश और अंत दे सकती है। आप नवीनतम डेटा निकाल सकते हैं, आप अंत से डेटा निकालना शुरू कर सकते हैं। यह हमारी पसंद है, लेकिन यह एक निश्चित अनुबंध है जिसे हम यहां निर्धारित करते हैं। यहाँ यह बफ़र (10, ओवरफ्लोस्ट्रेगेटी.ड्रॉपहेड) में है। यदि हम अब यह कार्यक्रम चलाते हैं,हमें इन चैनलों के ट्वीट अलग-अलग गति से प्राप्त होंगे। हमारे पास मालिक के नाम के साथ लगभग 10 ट्वीट्स हैं, समन्वय के साथ एक ट्वीट, क्योंकि हमने इस तरह की बैंडविड्थ निर्धारित की है। बेशक, मैं निर्देशांक के साथ कुछ करना चाहता हूं, अर्थात्, आप उन्हें कुछ सेवा में अपलोड करने का प्रयास कर सकते हैं जो हमें देगा, उदाहरण के लिए, मौसम कैसा है, अर्थात्, मौसम के आधार पर ट्वीट के लेखक किस मूड में थे। देखें कि अब इसे लागू करना कितना आसान होगा।क्या मूड में, मौसम के आधार पर, ट्वीट के लेखक थे। देखिए अब इसे लागू करना कितना आसान होगा।क्या मूड में, मौसम के आधार पर, ट्वीट का लेखक था। देखिए अब इसे लागू करना कितना आसान होगा।

 var graph = GraphDsl.Create(b => { var broadcast = b.Add(new Broadcast<ITweet>(2)); var merge = b.Add(new Merge<string>(2)); b.From(broadcast.Out(0)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.CreatedBy) .Throttle(10, TimeSpan.FromSeconds(1), 1, ThrottleMode.Shaping)) .Via(formatUser) .To(merge.In(0)); b.From(broadcast.Out(1)) .Via(Flow.Create<ITweet>().Select(tweet => tweet.Coordinates) .Buffer(10, OverflowStrategy.DropNew) .Throttle(1, TimeSpan.FromSeconds(1), 10, ThrottleMode.Shaping)) .Via(Flow.Create<ICoordinates>().SelectAsync(5, Utils.GetWeatherAsync)) .Via(formatTemperature) .To(merge.In(1)); return new FlowShape<ITweet, string>(broadcast.In, merge.Out); });

( स्रोत )

यहां हमारे पास दूसरा चैनल है, इसमें SelectAsync है, जिसमें हमें मौसम मिलता है। हम इसे केवल मौसम वितरण सेवा में नहीं भेजते हैं, हम यह भी कहते हैं कि इस कोड को 5 के समानांतर स्तर के साथ निष्पादित किया जाता है: इसका मतलब है कि 5 समानांतर धागे बनाए जाएंगे यदि यह सेवा काफी धीमी है जिसमें यह सेवा मौसम का अनुरोध करेगी। यह सेवा स्वयं यहां लागू की गई है, यह यह भी दिखाने के लिए समझ में आता है कि यह कोड कितना सरल दिखता है।

 public static async Task<decimal> GetWeatherAsync(ICoordinates coordinates) { var httpClient = new HttpClient(); var requestUrl = $"http://api.met.no/weatherapi/locationforecast/1.9/?lat={coordinates.Latitude};lon={coordinates.Latitude}"; var result = await httpClient.GetStringAsync(requestUrl); var doc = XDocument.Parse(result); var temp = doc.Root.Descendants("temperature").First().Attribute("value").Value; return decimal.Parse(temp); }

( स्रोत ) यहां

सब कुछ काफी सरल है। मुझे पहली वेब सेवा मिली, जो निर्देशांक द्वारा वर्तमान मौसम को दर्शाती है, यह कुछ प्रकार की नॉर्वेजियन मौसम-प्रदाता थी, यह सिर्फ HttpClient के माध्यम से यहां भेजा गया अनुरोध है, और एक्सएमएल से मौसम का डेटा प्राप्त किया जा सकता है जो मुझे मिलता है।

और अब, यदि हम अपना प्रदर्शन चलाते हैं , तो यदि यह सेवा अब चालू हो जाती है, तो तापमान डेटा देर से दिखाई देने लगेगा। 10 ट्वीट्स में लगभग एक बार हमारे पास मालिक के नाम के साथ 10 संदेश होते हैं, और एक बार हमें डिग्री सेल्सियस में तापमान मिलता है जहां ट्वीट लिखा गया था।

यह काफी प्रभावशाली है कि ऐसी प्रक्रिया का वर्णन करना कितना सरल है, जिसमें समानता के स्तर को इंगित करना शामिल है। ये केवल कुछ ही ब्लॉक हैं जिनका उपयोग अक्का धाराओं में किया जा सकता है, मैंने पहले ही कहा है कि उनमें से बहुत सारे हैं। संभावना है कि आप उनमें से कई का लाभ उठा सकते हैं वे काफी अधिक हैं।

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

अंतिम विचार

इस लेख को पढ़ने के बाद अक्का धाराओं के बारे में आप क्या विचार रखना चाहेंगे? DotNext 2017 मास्को में मैं एलेक्स थिससेन की प्रस्तुति पर थाएज़्योर फ़ंक्शंस के बारे में। एक अर्थ में, यह विचार है कि तैनाती के लिए कोड कैसे लिखें, कि हम मशीनों के कॉन्फ़िगरेशन पर ध्यान केंद्रित करने के बजाय, इस मशीन पर ऐसे प्रोग्राम स्थापित करते हैं जो इन सेवाओं पर बात करते हैं और यह डेटा प्राप्त करते हैं) , हम सीधे कार्यात्मक भाग और क्लाउड में इस कार्यात्मक हेलमेट पर ध्यान केंद्रित करते हैं। हम इस कोड को निष्पादित करने वाले मशीन नोड्स के बारे में बिल्कुल नहीं सोचते हैं, हम सोचते हैं कि हमारे कार्यों को एक दूसरे के साथ कैसे सहयोग करना चाहिए। एक समान उपमा एक प्रणाली के बीच खींची जा सकती है जिसे अभिनेता मॉडल का उपयोग करते हुए लिखा जाता है, लेकिन मैन्युअल रूप से और अक्का धाराएं, अर्थात्। हम भूल जाते हैं कि अभिनेताओं को मैन्युअल रूप से कैसे लिखा जाए और इसके बजाय पूरी प्रक्रिया विवरण पर ध्यान दें।परिदृश्यों के एक महत्वपूर्ण हिस्से में, हम सिस्टम की मापनीयता और गति को बनाए रखते हुए काफी उच्च स्तर पर बने रहने का प्रबंधन करते हैं।

चूंकि अक्का स्ट्रीम इस दृष्टिकोण का एकमात्र विकल्प नहीं है, जब आप सोचते हैं कि अपनी प्रक्रिया का अनुकरण कैसे किया जाए, तो इस बारे में सोचें कि क्या आप एक स्तर तक जा सकते हैं। उन सभी फायदों के लिए जो माइक्रोसॉफ़्ट हमें अखंड दृष्टिकोण की तुलना में देते हैं, कुछ निश्चित चिंताएँ हैं कि हम बहुत अधिक ध्यान केंद्रित करते हैं कि हम माइक्रो-सर्विसेस पर केंद्रित हों, हमें माइक्रो-मास्क मिलते हैं, और इस सब के लिए हम पूरे जंगल को नहीं देखते हैं। और अब अक्का धाराएं एक ऐसा तरीका है, जो अखंड स्तर पर लौटने के बिना, फिर भी प्रक्रिया के सामान्य विचार के स्तर पर वापस आ रहा है।

अंत में, मेरे पास आपके लिए एक छोटा सा गाना है, जो "अक्का स्ट्रीम रैप" कहे जाने वाले कुछ अंतर्निहित अक्का धाराओं के बारे में बात करता है। वीडियो के नीचे शब्द हैं, आप इसे चालू कर सकते हैं और साथ गा सकते हैं।

This is the Akka Stream.

This is the Source that feeds the Akka Stream.

This is the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Streams.

This is the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Throttle that speeds down the TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the Source that feeds the Akka Stream.

This is the Bidiflow that turns back the Throttle that speeds down the TakeWhile that pulls from the Drop that removes from the Zip that combines from the Balance that splits the FilterNot that selects from the Merge that collects from the Broadcast that forks the MapAsync that maps from the source that feeds the Akka Streams.

यह वह सिंक है जो बिडिफ्लो से भरा होता है जो थ्रॉटल को वापस ले जाता है जो टेकहाइल को गति देता है जो ड्रॉप से निकलने वाली ज़िप से खींचता है जो शेष राशि से जोड़ती है जो फ़िल्टरनोट को विभाजित करता है जो मर्ज से चुनता है जो प्रसारण से इकट्ठा होता है MapAsync के लिए कांटा स्रोत से नक्शे कि अक्का स्ट्रीम खिलाती है।

विज्ञापन का मिनट। यदि आपको रिपोर्ट पसंद आई और वह कुछ और पसंद करेगा, तो अगले DotNext 2018 मास्को 22-23 नवंबर को मास्को में आयोजित किया जाएगा , और यह आपके लिए कोई कम दिलचस्प नहीं हो सकता है। जुलाई की कीमत पर टिकट पाने के लिए जल्दी करें (1 अगस्त से, टिकटों की लागत बढ़ जाएगी)।

अक्का धाराएं केवल नश्वर के लिए