एक रूबी बैकग्राउंड प्रोसेसिंग सिस्टम बनाएं: स्क्रैच से सीखें

आज की पोस्ट में, हम मज़े के लिए एक सरल बैकग्राउंड प्रोसेसिंग सिस्टम लागू करने जा रहे हैं! हम रास्ते में कुछ चीजें सीख सकते हैं जैसे लोकप्रिय बैकग्राउंड प्रोसेसिंग सिस्टम जैसे साइडकिक के इंटर्नल्स में एक झलक। इस मज़े का उत्पाद किसी भी तरह से प्रोडक्शन उपयोग के लिए अभिप्रेत नहीं है।

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

एक कार्य को एनकैप्सुलेट करना

बैकग्राउंड प्रोसेसिंग में आने से पहले, आइए हाथ में कार्य करने के लिए एक सर्विस ऑब्जेक्ट बनाएं। हम टाइटल टैग की सामग्री निकालने के लिए OpenURI और Nokogiri का उपयोग करेंगे।

require 'open-uri' require 'nokogiri' class TitleExtractorService def call(url) document = Nokogiri::HTML(open(url)) title = document.css('html > head > title').first.content puts title.gsub(/[[:space:]]+/, ' ').strip rescue puts "Unable to find a title for #{url}" end end

सर्विस को कॉल करने से दिए गए URL का टाइटल प्रिंट होता है।

TitleExtractorService.new.call('https://appsignal.com') # AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir

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

module Magique module Worker def self.included(base) base.extend(ClassMethods) end module ClassMethods def perform_now(*args) new.perform(*args) end end def perform(*) raise NotImplementedError end end end

मॉड्यूल वर्कर इंस्टेंस में एक perform विधि और वर्कर क्लास में एक perform_now विधि जोड़ता है ताकि आह्वान थोड़ा बेहतर हो।

आइए मॉड्यूल को अपने सर्विस ऑब्जेक्ट में शामिल करें। जब हम इस पर हैं, तो आइए इसका नाम बदलकर TitleExtractorWorker कर दें और call विधि को perform में बदल दें।

class TitleExtractorWorker include Magique::Worker def perform(url) document = Nokogiri::HTML(open(url)) title = document.css('html > head > title').first.content puts title.gsub(/[[:space:]]+/, ' ').strip rescue puts "Unable to find a title for #{url}" end end

आह्वान का अभी भी वही परिणाम है, लेकिन यह थोड़ा स्पष्ट है कि क्या हो रहा है।

TitleExtractorWorker.perform_now('https://appsignal.com') # AppSignal: Application Performance Monitoring for Ruby on Rails and Elixir

एसिंक्रोनस प्रोसेसिंग लागू करना

अब जब हमारे पास टाइटल निष्कर्षण काम कर रहा है, तो हम पिछले रूबी मैजिक लेखों से सभी टाइटल्स ले सकते हैं। ऐसा करने के लिए, मान लें कि हमारे पास एक RUBYMAGIC कांस्टेंट है जिसमें पिछले लेखों के सभी URL की सूची है।

RUBYMAGIC.each do |url| TitleExtractorWorker.perform_now(url) end # Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog # Bindings and Lexical Scope in Ruby | AppSignal Blog # Building a Ruby C Extension From Scratch | AppSignal Blog # Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog # ...

हम पिछले लेखों के टाइटल्स प्राप्त करते हैं, लेकिन उन सभी को निकालने में थोड़ा समय लगता है। ऐसा इसलिए है क्योंकि हम अगले पर जाने से पहले प्रत्येक अनुरोध के पूरा होने की प्रतीक्षा करते हैं।

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

module Magique module Worker module ClassMethods def perform_async(*args) Thread.new { new.perform(*args) } end end end end

TitleExtractorWorker.perform_async(url) पर आह्वान बदलने के बाद, हमें लगभग एक साथ सभी टाइटल्स मिल जाते हैं। हालांकि, इसका मतलब यह भी है कि हम एक बार में रूबी मैजिक ब्लॉग पर 20 से अधिक कनेक्शन खोल रहे हैं। (आपके ब्लॉग के साथ गड़बड़ करने के लिए क्षमा करें, दोस्तों! 😅)

यदि आप अपने स्वयं के कार्यान्वयन के साथ अनुसरण कर रहे हैं और इसे लंबे समय तक चलने वाली प्रक्रिया (जैसे वेब सर्वर) के बाहर टेस्ट कर रहे हैं, तो अपने स्क्रिप्ट के अंत में loop { sleep 1 } जैसा कुछ जोड़ना न भूलें ताकि यह सुनिश्चित हो सके कि प्रक्रिया तुरंत समाप्त न हो।

कार्यों को कतारबद्ध करना

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

इस समस्या को हल करने का एक सामान्य तरीका उत्पादक/उपभोक्ता पैटर्न का उपयोग करना है। एक या अधिक उत्पादक कार्यों को एक कतार पर धकेलते हैं जबकि एक या अधिक उपभोक्ता कतार से कार्यों को लेते हैं और उन्हें प्रोसेस करते हैं।

एक कतार मूल रूप से तत्वों की एक सूची है। सिद्धांत रूप में, एक साधारण सरणी काम करेगी। हालांकि, चूंकि हम concurrency के साथ काम कर रहे हैं, हमें यह सुनिश्चित करने की आवश्यकता है कि केवल एक उत्पादक या उपभोक्ता एक समय में कतार तक पहुंच सके। यदि हम इस बारे में सावधान नहीं हैं, तो चीजें अराजकता में समाप्त हो जाएंगी — ठीक वैसे ही जैसे दो लोग एक साथ एक दरवाजे से निचोड़ने की कोशिश कर रहे हों।

इस समस्या को उत्पादक-उपभोक्ता समस्या के रूप में जाना जाता है और इसके कई समाधान हैं। सौभाग्य से, यह एक बहुत ही सामान्य समस्या है और रूबी एक उचित Queue कार्यान्वयन के साथ आती है जिसका उपयोग हम थ्रेड सिंक्रनाइज़ेशन के बारे में चिंता किए बिना कर सकते हैं।

इसका उपयोग करने के लिए, आइए सुनिश्चित करें कि उत्पादक और उपभोक्ता दोनों कतार तक पहुंच सकते हैं। हम अपने Magique मॉड्यूल में एक क्लास विधि जोड़कर और Queue का एक इंस्टेंस असाइन करके ऐसा करते हैं।

module Magique def self.backend @backend end def self.backend=(backend) @backend = backend end end Magique.backend = Queue.new

इसके बाद, हम अपने perform_async कार्यान्वयन को अपने स्वयं के नए थ्रेड बनाने के बजाय कतार पर एक कार्य को धकेलने के लिए बदलते हैं। एक कार्य को वर्कर क्लास के संदर्भ के साथ-साथ perform_async विधि को पास किए गए आर्गुमेंट्स सहित एक हैश के रूप में दर्शाया जाता है।

module Magique module Worker module ClassMethods def perform_async(*args) Magique.backend.push(worker: self, args: args) end end end end

इसके साथ, हम उत्पादक पक्ष के साथ काम कर चुके हैं। अब, आइए उपभोक्ता पक्ष पर एक नज़र डालें।

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

module Magique class Processor def self.start(concurrency = 1) concurrency.times { |n| new("Processor #{n}") } end def initialize(name) thread = Thread.new do loop do payload = Magique.backend.pop worker_class = payload[:worker] worker_class.new.perform(*payload[:args]) end end thread.name = name end end end

प्रोसेसिंग लूप के अलावा, हम Magique::Processor.start नामक एक सुविधा विधि जोड़ते हैं। यह हमें एक साथ कई प्रोसेसर्स को स्पिन अप करने की अनुमति देता है। जबकि थ्रेड का नाम देना वास्तव में आवश्यक नहीं है, यह हमें यह देखने की अनुमति देगा कि चीजें वास्तव में अपेक्षित रूप से काम कर रही हैं।

आइए अपने TitleExtractorWorker के आउटपुट को वर्तमान थ्रेड के नाम को शामिल करने के लिए समायोजित करें।

puts "[#{Thread.current.name}] #{title.gsub(/[[:space:]]+/, ' ').strip}"

हमारे बैकग्राउंड प्रोसेसिंग सेटअप का परीक्षण करने के लिए, हमें पहले अपने कार्यों को एन्क्यू करने से पहले प्रोसेसर्स का एक सेट स्पिन अप करना होगा।

Magique.backend = Queue.new Magique::Processor.start(5) RUBYMAGIC.each do |url| TitleExtractorWorker.perform_async(url) end # [Processor 3] Bindings and Lexical Scope in Ruby | AppSignal Blog # [Processor 4] Building a Ruby C Extension From Scratch | AppSignal Blog # [Processor 1] Unraveling Classes, Instances and Metaclasses in Ruby | AppSignal Blog # [Processor 0] Ruby's Hidden Gems, StringScanner | AppSignal Blog # [Processor 2] Fibers and Enumerators in Ruby: Turning Blocks Inside Out | AppSignal Blog # [Processor 4] Closures in Ruby: Blocks, Procs and Lambdas | AppSignal Blog # ...

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

कई प्रक्रियाओं और मशीनों तक विस्तार

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

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

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

आइए एक रेडिस-बैक्ड कतार बनाएं जो पहले उपयोग की गई Queue की तरह push और shift विधियों को लागू करती है।

require 'json' require 'redis' module Magique module Backend class Redis def initialize(connection = ::Redis.new) @connection = connection end def push(job) @connection.lpush('magique:queue', JSON.dump(job)) end def shift _queue, job = @connection.brpop('magique:queue') payload = JSON.parse(job, symbolize_names: true) payload[:worker] = Object.const_get(payload[:worker]) payload end end end end

चूंकि रूबी ऑब्जेक्ट्स के बारे में रेडिस कुछ नहीं जानता, हमें अपने कार्यों को रेडिस में स्टोर करने से पहले JSON में सीरियलाइज़ करना होगा lpush कमांड का उपयोग करके जो सूची के सामने एक तत्व जोड़ता है।

कतार से एक कार्य लाने के लिए, हम brpop कमांड का उपयोग कर रहे हैं, जो सूची से अंतिम तत्व प्राप्त करता है। यदि सूची खाली है, तो यह तब तक ब्लॉक करेगा जब तक एक नया तत्व उपलब्ध न हो जाए। जब कोई कार्य उपलब्ध न हो तो यह हमारे प्रोसेसर्स को रोकने का एक अच्छा तरीका है। अंत में, रेडिस से एक कार्य निकालने के बाद, हमें Object.const_get का उपयोग करके वर्कर के नाम के आधार पर वास्तविक रूबी क्लास को देखना होगा।

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

# ... Magique.backend = Magique::Backend::Redis.new RUBYMAGIC.each do |url| TitleExtractorWorker.perform_async(url) end

उपभोक्ता पक्ष पर, हम कुछ पंक्तियों जैसे इससे दूर हो सकते हैं:

# ... Magique.backend = Magique::Backend::Redis.new Magique::Processor.start(5) loop { sleep 1 }

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

जिम्मेदारी से आनंद लें और प्रोडक्शन में इसका उपयोग न करें

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

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

फिर भी, हमें इसे लिखने में मज़ा आया और आशा है कि आपने बैकग्राउंड प्रोसेसिंग सिस्टम के हुड के नीचे एक झलक का आनंद लिया। शायद आपने एक या दो चीजें भी लीं।