Node.js डमी के लिए धाराएँ या धाराओं के साथ कैसे काम करें

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



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

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

नोड js में 4 प्रकार की धाराएँ हैं:

• पठनीय - पढ़ने
• लिखने योग्य - लेखन
• डुप्लेक्स - पढ़ें और लिखें
• ट्रांसफ़ॉर्म - एक प्रकार का डुप्लेक्स स्ट्रीम जो डेटा को संशोधित कर सकता है

आप आधिकारिक वेबसाइट पर अधिक विस्तृत जानकारी पा सकते हैं, और अब अभ्यास करने के लिए आगे बढ़ते हैं।

सरल उदाहरण है

मुझे लगता है कि कई पहले से ही इसे महसूस किए बिना भी धाराओं का उपयोग कर चुके हैं। इस उदाहरण में, हम बस क्लाइंट को फाइल भेजते हैं।

// 1 - ( )      ,         const getFile = async (req, res, next) => { const fileStream = fs.createReadStream('path to file'); res.contentType('application/pdf'); fileStream.pipe(res); }; // 2 - (  )         const getFile = async (req, res, next) => { const file = fs.readFileSync('path to file'); res.contentType('application/pdf'); res.send(file); }; 

अंतर केवल इतना है कि पहले मामले में हम फ़ाइल का हिस्सा डाउनलोड करते हैं और इसे भेजते हैं, इस प्रकार, सर्वर की रैम को लोड नहीं करते हैं। दूसरे मामले में, हम तुरंत पूरी फाइल को रैम में लोड करते हैं और उसके बाद ही इसे भेजते हैं।

लेख में आगे हम प्रत्येक स्ट्रीम का अलग-अलग विश्लेषण करेंगे। आप इनहेरिटेंस का उपयोग करके या कंस्ट्रक्टर फ़ंक्शन का उपयोग करके एक स्ट्रीम बना सकते हैं।

 const { Readable } = require('stream'); // 1 -   const myReadable = new Readable(opt); // 2 -   class myReadable extends Readable { constructor(opt) { super(opt); } } 

सभी उदाहरणों में, मैं 2 विधि का उपयोग करूंगा।

पठनीय धारा

आइए देखें कि हम NodeJS में एक पठनीय स्ट्रीम कैसे बना सकते हैं।

 const { Readable } = require('stream'); class myReadable extends Readable { constructor(opt) { super(opt); } _read(size) {} } 

जैसा कि आप ऊपर के उदाहरण से देख सकते हैं, यह वर्ग मापदंडों के एक सेट को स्वीकार करता है। हम केवल उन पर विचार करेंगे जो पठनीय धारा की सामान्य समझ के लिए आवश्यक हैं, बाकी आप प्रलेखन में देख सकते हैं। हम HighWaterMark पैरामीटर और _read विधि में रुचि रखते हैं।

highWaterMark आंतरिक स्ट्रीम बफर के बाइट्स की अधिकतम संख्या है (डिफ़ॉल्ट रूप से 16kb) जिस पर पहुंचने से संसाधन से रीडिंग निलंबित है। पढ़ने को जारी रखने के लिए, हमें आंतरिक बफर को मुक्त करने की आवश्यकता है। हम इसे पाइप, फिर से शुरू करने के तरीकों, या डेटा इवेंट की सदस्यता के द्वारा कह सकते हैं।

_read एक निजी पद्धति का कार्यान्वयन है जिसे पठनीय वर्ग के आंतरिक तरीकों द्वारा बुलाया जाता है। इसे तब तक लगातार कहा जाता है जब तक कि डेटा का आकार हाईवाटरमार्क तक नहीं पहुंच जाता।

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

आइए अब स्थिति को स्पष्ट करने के लिए एक उदाहरण देखते हैं।

 class Counter extends Readable { constructor(opt) { super(opt); this._max = 1000; this._index = 0; } _read() { this._index += 1; if (this._index > this._max) { this.push(null); } else { const buf = Buffer.from(`${this._index}`, 'utf8'); console.log(`Added: ${this._index}. Could be added? `, this.push(buf)); } } } const counter = new Counter({ highWaterMark: 2 }); console.log(`Received: ${counter.read().toString()}`); 

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



यहाँ क्या हुआ? नया काउंटर ({highWaterMark: 2}) स्ट्रीम बनाते समय, हमने संकेत दिया कि हमारे आंतरिक बफर का आकार 2 बाइट्स होगा, अर्थात। 2 अक्षर (1 वर्ण = 1 बाइट) संग्रहीत कर सकते हैं। काउंटर.्रेड () को कॉल करने के बाद, स्ट्रीम पढ़ना शुरू कर देता है, आंतरिक बफर को '1' लिखता है और इसे वापस करता है। फिर उसने पढ़ना जारी रखा, '2' लिखा। जब यह '3' लिखता है, तो बफ़र पूरा भरा होगा, पठनीय.पुष झूठे वापस आ जाएगा, और धारा तब तक प्रतीक्षा करेगी जब तक कि आंतरिक बफर मुक्त नहीं हो जाता। क्योंकि हमारे उदाहरण में, बफर को मुक्त करने के लिए कोई तर्क नहीं है, स्क्रिप्ट समाप्त हो जाएगी।

जैसा कि पहले उल्लेख किया गया है, यह सुनिश्चित करने के लिए कि रीडिंग बाधित नहीं है, हमें आंतरिक बफर को लगातार साफ़ करने की आवश्यकता है। ऐसा करने के लिए, हम डेटा इवेंट की सदस्यता लेते हैं। अंतिम 2 पंक्तियों को निम्न कोड से बदलें।

 const counter = new Counter({ highWaterMark: 2 }); counter.on('data', chunk => { console.log(`Received: ${chunk.toString()}`); }); 

अब यदि हम इस उदाहरण को चलाते हैं, तो हम देखेंगे कि सब कुछ काम करना चाहिए जैसा कि और 1 से 1000 तक की संख्या कंसोल में प्रदर्शित होती है।

लिखित धारा

वास्तव में, यह एक पठनीय धारा के समान है, जिसका उद्देश्य केवल डेटा लिखना है।

 const { Writable } = require('stream'); class myWritable extends Writable { constructor(opt) { super(opt); } _write(chunk, encoding, callback) {} } 

यह पठनीय स्ट्रीम की तरह समान मापदंडों को स्वीकार करता है। हम highWaterMark और _write में रुचि रखते हैं।

_write एक निजी विधि है जिसे डेटा का एक टुकड़ा लिखने के लिए लिखने योग्य वर्ग के आंतरिक तरीकों द्वारा कहा जाता है। यह 3 पैरामीटर लेता है: चंक (डेटा का हिस्सा), एन्कोडिंग (अगर चंक एक स्ट्रिंग है), कॉलबैक (एक फ़ंक्शन जिसे एक सफल या असफल लेखन के बाद कहा जाता है)।

highWaterMark आंतरिक स्ट्रीम बफर (डिफ़ॉल्ट रूप से 16kb) की बाइट्स की अधिकतम संख्या है, जिस पर पहुंचने पर स्ट्रीम। राइट झूठे वापस आने लगेगा।

आइए एक काउंटर के साथ पिछले उदाहरण को फिर से लिखें।

 const { Writable } = require('stream'); class Counter extends Writable { _write(chunk, encoding, callback) { console.log(chunk.toString()); callback(); } } const counter = new Counter({ highWaterMark: 2 }); for (let i = 1; i < 1000; i += 1) { counter.write(Buffer.from(`${i}`, 'utf8')); } 

वास्तव में, सब कुछ सरल है, लेकिन एक दिलचस्प बारीकियों को याद रखने योग्य है! नया काउंटर ({highWaterMark: 2}) स्ट्रीम बनाते समय, हमने संकेत दिया कि हमारे आंतरिक बफर का आकार 2 बाइट्स होगा, अर्थात। 2 अक्षर (1 वर्ण = 1 बाइट) संग्रहीत कर सकते हैं। जब काउंटर दस तक पहुंचता है, तो बफ़र लिखने के लिए प्रत्येक कॉल के साथ भरा जाएगा, क्रमशः, यदि रिकॉर्डिंग एक धीमी स्रोत में किया गया था, तो कॉलिंग लिखने के अन्य सभी डेटा रैम में संग्रहीत किए जाएंगे, जिससे यह अतिप्रवाह हो सकता है (इस उदाहरण में, बेशक यह महत्वपूर्ण नहीं है, क्योंकि हमारा बफर 2 बाइट्स है, लेकिन बड़ी फ़ाइलों के साथ आपको यह याद रखना होगा)। जब ऐसी स्थिति उत्पन्न होती है, तो हमें तब तक इंतजार करना होगा जब तक कि धारा डेटा के वर्तमान हिस्से को नहीं लिखती है, आंतरिक बफर (ड्रेन इवेंट को ट्रिगर करता है) को रिलीज़ करती है, और फिर हम रिकॉर्डिंग डेटा को फिर से शुरू कर सकते हैं। आइए हमारे उदाहरण को फिर से लिखें।

 const { Writable } = require('stream'); const { once } = require('events'); class Counter extends Writable { _write(chunk, encoding, callback) { console.log(chunk.toString()); callback(); } } const counter = new Counter({ highWaterMark: 2 }); (async () => { for (let i = 1; i < 1000; i += 1) { const canWrite = counter.write(Buffer.from(`${i}`, 'utf8')); console.log(`Can we write bunch of data? ${canWrite}`); if (!canWrite) { await events.once(counter, 'drain'); console.log('drain event fired.'); } } })(); 

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

पहली नज़र में, यह एक अनावश्यक कार्रवाई की तरह लग सकता है, लेकिन बड़ी मात्रा में डेटा के साथ काम करते समय, उदाहरण के लिए 10GB से अधिक वजन वाली फाइलें, ऐसा करने के लिए भूल जाने पर, आप एक स्मृति रिसाव का सामना कर सकते हैं।

द्वैध धारा

यह पठनीय और लिखने योग्य धाराओं को जोड़ती है, अर्थात, हमें दो _read और _write विधियों के कार्यान्वयन को लिखना होगा।

 const { Duplex } = require('stream'); class myDuplex extends Duplex { constructor(opt) { super(opt); } _read(size) {} _write(chunk, encoding, callback) {} } 

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

 const { Duplex } = require('stream'); const events = require('events'); class Counter extends Duplex { constructor(opt) { super(opt); this._max = 1000; this._index = 0; } _read() { this._index += 1; if (this._index > this._max) { this.push(null); } else { const buf = Buffer.from(`${this._index}`, 'utf8'); this.push(buf); } } _write(chunk, encoding, callback) { console.log(chunk.toString()); callback(); } } const counter = new Counter({ readableHighWaterMark: 2, writableHighWaterMark: 2 }); (async () => { let chunk = counter.read(); while (chunk !== null) { const canWrite = counter.write(chunk); console.log(`Can we write bunch of data? ${canWrite}`); if (!canWrite) { await events.once(counter, 'drain'); console.log('drain event fired.'); } chunk = counter.read(); } })(); 

मुझे लगता है कि इस कोड को स्पष्टीकरण की आवश्यकता नहीं है, क्योंकि यह पहले की तरह ही है, केवल एक वर्ग में।

ट्रांसफ़ॉर्म स्ट्रीम

यह धारा एक द्वैध धारा है। यह डेटा के एक हिस्से को बदलने और इसे नीचे श्रृंखला को भेजने के लिए आवश्यक है। इसे बाकी स्ट्रीम की तरह ही लागू किया जा सकता है।

 const { Transform } = require('stream'); class myTransform extends Transform { _ transform(chunk, encoding, callback) {} } 

हम _transform विधि में रुचि रखते हैं।

_transform एक निजी विधि है जिसे ट्रांसफ़ॉर्म वर्ग के आंतरिक तरीकों से डेटा का एक हिस्सा बदलने के लिए कहा जाता है। यह 3 पैरामीटर लेता है: चंक (डेटा का हिस्सा), एन्कोडिंग (अगर चंक एक स्ट्रिंग है), कॉलबैक (एक फ़ंक्शन जो एक सफल या असफल लेखन के बाद कहा जाता है)।

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

stream.pipe - इस मेथड का उपयोग रीडेबल स्ट्रीम को Writable स्ट्रीम से जोड़ने के लिए किया जाता है, साथ ही स्ट्रीम चेन बनाने के लिए भी किया जाता है। इसका मतलब यह है कि हम डेटा का हिस्सा पढ़ सकते हैं और इसे प्रसंस्करण के लिए अगली स्ट्रीम में स्थानांतरित कर सकते हैं, और फिर अगले के लिए, आदि।

आइए एक ट्रांसफ़ॉर्म स्ट्रीम लिखें जो डेटा के प्रत्येक टुकड़े की शुरुआत और अंत में * कैरेक्टर जोड़ेगी।

 class CounterReader extends Readable { constructor(opt) { super(opt); this._max = 1000; this._index = 0; } _read() { this._index += 1; if (this._index > this._max) { this.push(null); } else { const buf = Buffer.from(`${this._index}`, 'utf8'); this.push(buf); } } } class CounterWriter extends Writable { _write(chunk, encoding, callback) { console.log(chunk.toString()); callback(); } } class CounterTransform extends Transform { _transform(chunk, encoding, callback) { try { const resultString = `*${chunk.toString('utf8')}*`; callback(null, resultString); } catch (err) { callback(err); } } } const counterReader = new CounterReader({ highWaterMark: 2 }); const counterWriter = new CounterWriter({ highWaterMark: 2 }); const counterTransform = new CounterTransform({ highWaterMark: 2 }); counterReader.pipe(counterTransform).pipe(counterWriter); 

इस उदाहरण में, मैंने पिछले उदाहरणों से पठनीय और लिखने योग्य धाराओं का उपयोग किया, और ट्रांसफ़ॉर्म भी जोड़ा। जैसा कि आप देख सकते हैं, यह बहुत सरल निकला।

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