Node.JS 10.5.0 में थ्रेड के साथ खेलना

शुभ दिन

मेरे काम पर, मेरे और सहयोगी के बीच Node.JS के नए संस्करण में थ्रेड्स के बारे में विवाद उत्पन्न हुआ और उन्हें सिंक्रनाइज़ करने की आवश्यकता हुई। इसके साथ शुरू करने के लिए, हमने समानांतर में फ़ाइल में लाइनों को लिखने का कार्य चुनने का फैसला किया। कार्यकर्ता के साथ विषय गर्म है, कृपया, बिल्ली के नीचे।

धाराओं के बारे में थोड़ा खुद। वे Node.JS 10.5.0 में प्रायोगिक तकनीक हैं, और "कार्यकर्ता_थ्रेड्स" मॉड्यूल तक पहुंचने के लिए, आपको "--experimental-कार्यकर्ता" ध्वज के साथ हमारे Node.JS एप्लिकेशन को चलाने की आवश्यकता है। मैंने इस झंडे को स्टार्ट स्क्रिप्ट में package.json फ़ाइल में पंजीकृत किया है:

{ "name": "worker-test", "version": "1.0.0", "description": "", "main": "app.js", "scripts": { "start": "node --max-old-space-size=4096 --experimental-worker app.js " }, "author": "", "license": "ISC" } 

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

दरअसल, कोड।

मुख्य app.js फ़ाइल प्रविष्टि बिंदु है।

 const { Worker } = require('worker_threads'); const path = require('path'); const WORKERS_NUMBER = 100; console.log('Hello from main!'); for (var i = 1; i <= WORKERS_NUMBER ; i++) { const w = new Worker(path.join(__dirname, './writer-worker-app/app.js'), { workerData: { id: i } }); } 

यहां हम वर्कर क्लास का उपयोग करके चाइल्ड स्ट्रीम बनाते हैं और स्ट्रीम के लिए स्टार्ट फाइल का रास्ता निर्दिष्ट करते हैं './writer-worker-app/app.js'। स्ट्रीम बनाते समय, हम स्वयं-लिखित पहचानकर्ता को कार्यकर्ता डेटा के रूप में स्थानांतरित करते हैं।

स्ट्रीम के लिए फ़ाइल प्रारंभ करें ।/writer-worker-app/app.js:

 const { workerData, parentPort } = require('worker_threads'); const logger = require('./logger'); const id = workerData.id; console.log(`Worker ${id} initializad.`); while (true) { sendMessage(); } function sendMessage() { logger.log(`Hello from worker number ${workerData.id}\r\n`); } 

खैर, सबसे सरल लकड़हारा वर्ग: ./writer-worker-app/logger.js

 const fs = require('fs'); function log(message) { return fs.appendFileSync('./my-file.txt', message); } module.exports = { log }; 

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

Hello from worker number 14
Hello from worker number 3
Hello from worker number 9
Hello from worker number 15
Hello from worker number 2
Hello from worker number 4
Hello from worker number 7
Hello from worker number 6
Hello from worker number 1
Hello from worker number 11

एक अद्भुत प्रयोग, नोदा के लिए एक और छोटी जीत :-) मेरी धारणा है कि सभी सिंक्रनाइज़ेशन नोदा के I / O स्तर पर होता है, लेकिन मुझे टिप्पणियों में सही विकल्प जानकर खुशी होगी। बस मामले में, हमने fs.appendFileSync का उपयोग करके काम की जाँच की, लेकिन fs.createWriteStream और stream.write विधि।

नतीजा वही निकला।

लेकिन हम वहां नहीं रुके।

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

ऐसा करने के लिए, मैंने एक और धागा प्रबंधक पेश किया। मुख्य चीज से प्राप्त करना संभव था, लेकिन मैं इन अलग-थलग श्रमिकों को बनाने और संदेशों के माध्यम से संचार बनाने के लिए बहुत खुश हूं। स्ट्रीम-मैनेजर का कार्यान्वयन लिखना शुरू करने से पहले, आपको उसके और लेखकों-श्रमिकों के बीच एक संचार चैनल बनाने की आवश्यकता है। इसके लिए MessageChannel क्लास का इस्तेमाल किया गया था। इस वर्ग के उदाहरणों में दो फ़ील्ड हैं: port1 और port2 , जिनमें से प्रत्येक .on ( 'संदेश') और .postMessage () विधियों का उपयोग करके दूसरे को संदेश सुन और भेज सकते हैं। यह वर्ग थ्रेड्स के बीच संचार के लिए "कार्यकर्ता_थ्रेड्स" मॉड्यूल के हिस्से के रूप में बनाया गया था, क्योंकि आमतौर पर जब किसी वस्तु को स्थानांतरित किया जाता है, तो इसे बस क्लोन किया जाता है, और एक अलग थ्रेड निष्पादन वातावरण में यह बेकार हो जाएगा।

2 प्रवाह के बीच संचार के लिए, हमें सभी को एक पोर्ट देना होगा।

एक दिलचस्प तथ्य : 10.5.0 पर श्रमिक के कंस्ट्रक्टर के माध्यम से पोर्ट को पारित करना असंभव है , आपको इसे केवल कार्यकर्ता.पोस्टमैसेज () के माध्यम से करने की आवश्यकता है, और ट्रांसफरलिस्ट पैरामीटर में पोर्ट को निर्दिष्ट करना सुनिश्चित करें!

थ्रेड मैनेजर स्वयं अपने पहचानकर्ताओं के आरोही क्रम में लेखक के थ्रेड्स को कमांड भेजेगा, और यह सफल ऑपरेशन के बारे में लेखक से प्रतिक्रिया प्राप्त करने के बाद ही अगली कमांड भेजेगा।

अंडर-यूएमएल एप्लिकेशन आरेख:


हमारी उत्परिवर्तित मुख्य ./app.js फ़ाइल:

 const { Worker, MessageChannel } = require('worker_threads'); const path = require('path'); const WORKERS_NUMBER = 100; console.log('Main app initialized and started.'); const workersMeta = []; for (var i = 1; i <= WORKERS_NUMBER; i++) { const channel = new MessageChannel(); const worker = new Worker(path.join(__dirname, './writer-worker-app/app.js'), { workerData: { id: i } }); workersMeta.push({ id: i, worker, channel }); } workersMeta.forEach(({ worker, channel }) => { worker.postMessage({ orchestratorPort: channel.port1 }, [channel.port1]); }) setTimeout(() => { const orchestrator = new Worker(path.join(__dirname, './orchestrator-worker-app/app.js')); const orchestratorData = workersMeta.map((meta) => ({ id: meta.id, port: meta.channel.port2 })); orchestrator.postMessage({ workerPorts: orchestratorData }, orchestratorData.map(w => w.port)); console.log('All worker threads have been initialized'); }, WORKERS_NUMBER * 10); 

यहां हम पहले श्रमिकों का निर्माण करते हैं, फिर हम प्रत्येक पोर्ट को प्रबंधक के साथ संचार के लिए एक पोर्ट भेजते हैं (और केवल इस तरह से, कंस्ट्रक्टर के माध्यम से ऐसा करना असंभव है)।

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

हम लेखक थ्रेड के व्यवहार को भी बदल देंगे ताकि यह केवल एक संदेश भेजता है जब इसे बताया जाता है, और परिणाम भी देता है जब लेखन ऑपरेशन पूरा हो जाता है:
./writer-worer-app/app.js

 const { workerData, parentPort } = require('worker_threads'); const logger = require('./logger'); const id = workerData.id; console.log(`Worker ${id} initializad.`); parentPort.on('message', value => { const orchestratorPort = value.orchestratorPort; orchestratorPort.on('message', data => { if (data.command == 'write') { console.log(`Worker ${id} received write command`); sendMessage(); sendResult(orchestratorPort); } }); console.log(`Worker ${id} started.`); }); function sendMessage() { logger.log(`Hello from worker number ${workerData.id}\r\n`); } function sendResult(port) { port.postMessage({ id, status: 'completed' }); } 

हमने मूल स्ट्रीम के संदेश से सही तरीके से आरंभ किया, स्ट्रीम मैनेजर का चैनल होने लगा, जब हम कमांड प्राप्त करते हैं, तो हम पहले फाइल पर लिखते हैं, फिर हम परिणाम भेजते हैं। यह ध्यान दिया जाना चाहिए कि फ़ाइल को सिंक्रोनाइज़ किया गया है, इसलिए sendResult () को तुरंत SendMessage () कहा जाता है।

वह सब कुछ हमारे स्मार्ट मैनेजर के कार्यान्वयन को लिखना है
./orchestrator-worker-app/app.js:

 const { parentPort } = require('worker_threads'); console.log('Orchestrator initialized.') let workerPorts; parentPort.on('message', (value) => { workerPorts = value.workerPorts; workerPorts.forEach(wp => wp.port.on('message', handleResponse)); console.log('Orchestrator started.'); sendCommand(workerPorts[0]); }); function handleResponse(status) { const responseWorkerId = status.id; let nextWorker = workerPorts.find(wp => wp.id == responseWorkerId + 1); if (!nextWorker) { nextWorker = workerPorts[0]; } sendCommand(nextWorker); } function sendCommand(worker) { worker.port.postMessage({ command: 'write' }); } 

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

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

सभी स्रोत कोड यहां देखे जा सकते हैं ।

अद्यतन

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

करने के लिए:
1) चर लंबाई के संदेश जोड़ें या लकड़हारे की कॉल की गिनती - धन्यवाद FANAT1242
2) एक बेंचमार्क जोड़ें, पहले और दूसरे संस्करणों के काम की तुलना करें (उदाहरण के लिए 10 सेकंड में कितनी लाइनें लिखी जाएंगी)

अद्यतन २

1) लॉगिंग कोड बदल दिया गया है: अब प्रत्येक संदेश की एक अलग लंबाई है।
2) लेखक-कार्यकर्ता-ऐप / app.old.js को बदल दिया गया है: प्रत्येक धागा 1000 बार लिखता है, फिर समाप्त होता है।

यह FANAT1242 उपयोगकर्ता के विचारों का परीक्षण करने के लिए किया गया था। सभी समान संदेश एक-दूसरे को फिर से नहीं लिखते हैं, फ़ाइल में लाइनें बिल्कुल 1000 * एन थ्रेड हैं।