Node.js ماهیت ناهمزمان و رویداد محور است. در نتیجه، در انجام وظایف محدود به I/O بسیار خوب است. اگر روی برنامه‌ای کار می‌کنید که عملیات I/O را انجام می‌دهد، می‌توانید از جریان‌های موجود در Node.js استفاده کنید. بنابراین، بیایید جریان‌ها را با جزئیات بررسی کنیم و بفهمیم که چگونه می‌توانند I/O را ساده‌سازی کنند.

خوراکی های کلیدی

جریان‌های Node.js، که ناهمزمان و رویداد محور هستند، می‌توانند عملیات ورودی/خروجی را با مدیریت کارآمد داده‌ها در قطعات کوچک‌تر و قابل مدیریت ساده‌تر کنند.

جریان ها را می توان به عنوان خواندنی، نوشتنی، دوبلکس (هم قابل خواندن و هم قابل نوشتن) یا تبدیل (تغییر داده ها در حین عبور) طبقه بندی کرد.

تابع ' pipe() ' یک ابزار مفید در جریان های Node.js است که اجازه می دهد تا داده ها از یک منبع خوانده شوند و بدون مدیریت دستی جریان داده به مقصد نوشته شوند.

Node.js مدرن ابزارهایی مانند " stream.pipeline() " و " stream.finished() " را همراه با API های مبتنی بر Promise برای مدیریت بهتر خطا و کنترل جریان ارائه می دهد.

جریان‌ها را می‌توان با الگوهای async/wait برای کد تمیزتر و قابل نگهداری‌تر استفاده کرد.

استریم ها چیست؟

جریان‌ها در Node.js از لوله‌های یونیکس الهام گرفته شده‌اند و مکانیزمی برای خواندن داده‌ها از یک منبع و انتقال آن‌ها به مقصد به صورت جریانی ارائه می‌دهند.

به زبان ساده، استریم چیزی جز یک EventEmitter نیست و برخی از روش‌های خاص را پیاده‌سازی می‌کند. بسته به روش های پیاده سازی شده، یک جریان به خواندن، نوشتن، دوبلکس یا تبدیل می شود. جریان های قابل خواندن به شما امکان می دهند داده ها را از یک منبع بخوانید در حالی که جریان های قابل نوشتن به شما امکان می دهند داده ها را در مقصد بنویسید.

اگر قبلاً با Node.js کار کرده اید، ممکن است با جریان هایی روبرو شده باشید. به عنوان مثال، در سرور HTTP مبتنی بر Node.js، request یک جریان قابل خواندن و response یک جریان قابل نوشتن است. ممکن است از ماژول fs استفاده کرده باشید که به شما امکان می دهد با جریان فایل های قابل خواندن و نوشتن کار کنید.

بیایید انواع مختلف جریان ها را درک کنیم. در این مقاله، ما در درجه اول بر روی جریان های قابل خواندن و نوشتن تمرکز می کنیم، اما به طور خلاصه به جریان های Duplex و Transform نیز می پردازیم.

جریان خواندنی

یک جریان قابل خواندن به شما امکان می دهد داده ها را از یک منبع بخوانید. منبع می تواند هر چیزی باشد. این می تواند یک فایل ساده در سیستم فایل شما، یک بافر در حافظه یا حتی یک جریان دیگر باشد. از آنجایی که جریان ها EventEmitters هستند، چندین رویداد را در نقاط مختلف منتشر می کنند. ما از این رویدادها برای کار با جریان ها استفاده خواهیم کرد.

خواندن از جریان

بهترین راه برای خواندن داده ها از یک جریان، گوش دادن به رویداد data و پیوست کردن یک تماس برگشتی است. هنگامی که تکه ای از داده در دسترس است، جریان قابل خواندن یک رویداد data را منتشر می کند و پاسخ تماس شما اجرا می شود. به قطعه زیر دقت کنید:

const fs = require ( 'fs' ) ; const readableStream = fs . createReadStream ( 'file.txt' ) ; let data = '' ;
readableStream . on ( 'data' , function ( chunk ) { data += chunk ; } ) ;
readableStream . on ( 'end' , function ( ) { console . log ( data ) ; } ) ;

readableStream . on ( 'error' , ( err ) => { console . error ( 'Error reading stream:' , err ) ; } ) ;

فراخوانی تابع fs.createReadStream() یک جریان قابل خواندن به شما می دهد. در ابتدا جریان در حالت ایستا قرار دارد. به محض اینکه به رویداد data گوش می دهید و یک تماس برگشتی را پیوست می کنید، شروع به جریان می کند. پس از آن، تکه‌هایی از داده‌ها خوانده می‌شوند و به تماس شما ارسال می‌شوند. پیاده‌کننده جریان تصمیم می‌گیرد که رویداد data هر چند وقت یک‌بار منتشر شود. به عنوان مثال، یک درخواست HTTP ممکن است یک رویداد data را پس از خواندن چند کیلوبایت داده منتشر کند. هنگامی که در حال خواندن داده ها از یک فایل هستید، ممکن است تصمیم بگیرید که پس از خواندن یک خط، رویداد data را منتشر کنید.

هنگامی که داده دیگری برای خواندن وجود ندارد (پایان به پایان می رسد)، جریان یک رویداد end منتشر می کند. در قطعه بالا، ما به این رویداد گوش می دهیم تا زمانی که به پایان رسید، مطلع شویم.

با ویژگی‌های مدرن ECMAScript، می‌توانیم این را با استفاده از async/wait بازنویسی کنیم:

 const fs = require('fs'); const { Readable } = require('stream'); const { promisify } = require('util');
// Convert stream.on('end') to a Promise const streamToString = async (stream) => { const chunks = [];  
 for await (const chunk of stream) { chunks.push(typeof chunk === 'string' ? chunk : chunk.toString()); }  
 return chunks.join(''); };
async function readFile() { try { const readableStream = fs.createReadStream('file.txt'); const content = await streamToString(readableStream); console.log(content); } catch (err) { console.error('Error reading file:', err); } }
readFile();

در اینجا، ما از چندین ویژگی جدیدتر جاوا اسکریپت استفاده می کنیم:

حلقه for await...of به ما امکان می دهد تا روی تکرارهای غیر همگام (مانند جریان ها در Node.js) تکرار کنیم.

ما در حال ایجاد یک تابع کمکی streamToString هستیم که تمام تکه‌های یک جریان را جمع‌آوری می‌کند و یک Promise را برمی‌گرداند که به رشته کامل حل می‌شود.

برای مدیریت صحیح خطا، همه چیز را در یک بلوک try/catch قرار می دهیم

این رویکرد خطی‌تر و راحت‌تر از رویکرد مبتنی بر رویداد است

اکنون دو حالت وجود دارد که یک جریان خواندنی می‌تواند در آن کار کند :

1. حالت جریان - داده ها به طور خودکار خوانده می شوند و در سریع ترین زمان ممکن از طریق رویدادها ارائه می شوند
2. حالت مکث - شما باید صریحاً read() را فراخوانی کنید تا تکه های داده را به طور مکرر دریافت کنید تا زمانی که هر تکه داده خوانده شود.

 const fs = require ( 'fs' ) ; const readableStream = fs . createReadStream ( 'file.txt' ) ; let data = '' ; let chunk ;
readableStream . on ( 'readable' , function ( ) { while ( ( chunk = readableStream . read ( ) ) != null ) { data += chunk ; } } ) ;
readableStream . on ( 'end' , function ( ) { console . log ( data ) ; } ) ;

تابع read() برخی از داده ها را از بافر داخلی می خواند و آن را برمی گرداند. وقتی چیزی برای خواندن وجود ندارد، آن را null برمی‌گرداند. بنابراین، در حلقه while ما null را بررسی می کنیم و حلقه را خاتمه می دهیم. توجه داشته باشید که رویداد readable زمانی منتشر می‌شود که تکه‌ای از داده‌ها را بتوان از جریان خواند.

تنظیم رمزگذاری

به طور پیش‌فرض داده‌هایی که از یک جریان می‌خوانید یک شی Buffer است. اگر رشته ها را می خوانید ممکن است برای شما مناسب نباشد. بنابراین، می‌توانید با فراخوانی Readable.setEncoding() کدگذاری را روی جریان تنظیم کنید، همانطور که در زیر نشان داده شده است.

 const fs = require ( 'fs' ) ; const readableStream = fs . createReadStream ( 'file.txt' ) ; let data = '' ;
readableStream . setEncoding ( 'utf8' ) ;
readableStream . on ( 'data' , function ( chunk ) { data += chunk ; } ) ;
readableStream . on ( 'end' , function ( ) { console . log ( data ) ; } ) ;

در قطعه بالا کدگذاری را روی utf8 قرار دادیم. در نتیجه، داده ها به صورت utf8 تفسیر می شوند و به عنوان رشته به callback شما ارسال می شوند.

لوله کشی

Piping یک مکانیسم عالی است که در آن می‌توانید داده‌ها را از منبع بخوانید و بدون مدیریت جریان خود، به مقصد بنویسید. به قطعه زیر دقت کنید:

 const fs = require ( 'fs' ) ; const readableStream = fs . createReadStream ( 'file1.txt' ) ; const writableStream = fs . createWriteStream ( 'file2.txt' ) ;
readableStream . pipe ( writableStream ) ;

قطعه بالا از تابع pipe() برای نوشتن محتوای file1 به file2 استفاده می کند. از آنجایی که pipe() جریان داده را برای شما مدیریت می کند، نباید نگران جریان داده آهسته یا سریع باشید. این باعث می شود pipe() ابزاری منظم برای خواندن و نوشتن داده ها باشد. همچنین باید توجه داشته باشید که pipe() جریان مقصد را برمی گرداند. بنابراین، شما به راحتی می توانید از این برای زنجیره ای کردن چندین جریان با هم استفاده کنید. بیایید ببینیم چگونه!

با این حال، یکی از محدودیت های pipe() این است که مدیریت خطا را به خوبی ارائه نمی دهد. اینجاست که Node.js مدرن ابزارهای بهتری را ارائه می دهد:

 const fs = require ( 'fs' ) ; const { pipeline } = require ( 'stream' ) ; const { promisify } = require ( 'util' ) ;
const pipelineAsync = promisify ( pipeline ) ;
async function copyFile ( ) { try { const readableStream = fs . createReadStream ( 'file1.txt' ) ; const writableStream = fs . createWriteStream ( 'file2.txt' ) ;    
await pipelineAsync ( readableStream , writableStream ) ; console . log ( 'File copied successfully' ) ; } catch ( err ) { console . error ( 'Pipeline failed:' , err ) ; } }
copyFile ( ) ;

اینجا:

ما از تابع pipeline از ماژول جریان استفاده می کنیم که به طور خودکار خطاها و پاکسازی منابع را کنترل می کند.

ما خط لوله مبتنی بر callback را با استفاده از promisify به Promise تبدیل می کنیم

سپس می‌توانیم از async/wait برای یک جریان تمیزتر استفاده کنیم.

همه خطاها به درستی در یک بلوک تلاش/گرفتن ثبت می‌شوند.

اگر هر جریانی در خط لوله خطا صادر کند، خط لوله به طور خودکار همه جریان‌ها را از بین می‌برد و تماس با خطا را فراخوانی می‌کند.

زنجیر زدن

فرض کنید یک آرشیو دارید و می خواهید آن را از حالت فشرده خارج کنید. راه های مختلفی برای رسیدن به این هدف وجود دارد. اما ساده ترین و تمیزترین راه استفاده از لوله کشی و زنجیر کشی است. به قطعه زیر نگاهی بیندازید:

 const fs = require ( 'fs' ) ; const zlib = require ( 'zlib' ) ;
fs . createReadStream ( 'input.txt.gz' ) . pipe ( zlib . createGunzip ( ) ) . pipe ( fs . createWriteStream ( 'output.txt' ) ) ;

ابتدا یک جریان ساده قابل خواندن از فایل input.txt.gz ایجاد می کنیم. سپس، این استریم را به یک جریان دیگر zlib.createGunzip() لوله می کنیم تا محتوا را از حالت gzip خارج کنیم. در نهایت، از آنجایی که جریان‌ها را می‌توان زنجیره‌ای کرد، یک جریان قابل نوشتن اضافه می‌کنیم تا محتوای بدون gzip را در فایل بنویسیم.

یک رویکرد قوی تر با استفاده از خط لوله:

 const fs = require ( 'fs' ) ; const zlib = require ( 'zlib' ) ; const { pipeline } = require ( 'stream' ) ;
pipeline ( fs . createReadStream ( 'input.txt.gz' ) , zlib . createGunzip ( ) , fs . createWriteStream ( 'output.txt' ) , ( err ) => { if ( err ) { console . error ( 'Pipeline failed:' , err ) ; } else { console . log ( 'Pipeline succeeded' ) ; } } ) ;

در اینجا ما از خط لوله با چندین جریان استفاده می کنیم:

برخلاف pipe() که به درستی خطاها را ارسال نمی کند، خط لوله خطاها را از هر جریانی در زنجیره کنترل می کند.

اگر هر جریانی در خط لوله شکست بخورد (مثلاً اگر فایل وجود نداشته باشد یا محتوا gzip معتبر نباشد)، callback خطا را دریافت می‌کند.

Pipeline به طور خودکار منابع را با از بین بردن تمام جریان ها در صورت وجود خطا در جریان پاک می کند.

آخرین آرگومان یک callback است که به ما می گوید آیا عملیات موفقیت آمیز بوده یا شکست خورده است.

روش های اضافی

ما برخی از مفاهیم مهم را در جریان های خواندنی مورد بحث قرار دادیم. در اینجا چند روش پخش جریانی دیگر وجود دارد که باید بدانید:

Readable.pause() – این روش جریان را متوقف می کند. اگر جریان در حال حاضر جریان داشته باشد، دیگر رویدادهای data را منتشر نمی کند. داده ها در بافر نگهداری می شوند. اگر این را روی یک جریان ثابت (غیر جاری) صدا کنید، هیچ اثری وجود ندارد و جریان متوقف می‌شود.

Readable.resume() – یک جریان متوقف شده را از سر می گیرد.

readable.unpipe() – این جریان های مقصد را از مقصد لوله حذف می کند. اگر آرگومان ارسال شود، جریان قابل خواندن را از لوله کشی به جریان مقصد خاص متوقف می کند. در غیر این صورت، تمام جریان های مقصد حذف می شوند.

جریان های قابل نوشتن

جریان های قابل نوشتن به شما امکان می دهند داده ها را در یک مقصد بنویسید. مانند جریان‌های قابل خواندن، اینها نیز EventEmitters هستند و رویدادهای مختلفی را در نقاط مختلف منتشر می‌کنند. بیایید روش‌ها و رویدادهای مختلف موجود در جریان‌های قابل نوشتن را ببینیم.

نوشتن به استریمز

برای نوشتن داده‌ها در یک جریان قابل نوشتن، باید write() در نمونه استریم فراخوانی کنید. قطعه زیر این تکنیک را نشان می دهد.

 const fs = require ( 'fs' ) ; const readableStream = fs . createReadStream ( 'file1.txt' ) ; const writableStream = fs . createWriteStream ( 'file2.txt' ) ;
readableStream . setEncoding ( 'utf8' ) ;
readableStream . on ( 'data' , function ( chunk ) { writableStream . write ( chunk ) ; } ) ;

کد بالا ساده است. به سادگی تکه های داده را از یک جریان ورودی می خواند و با استفاده از write() به مقصد می نویسد. این تابع یک مقدار بولی برمی گرداند که نشان می دهد آیا عملیات موفقیت آمیز بوده است.

مقدار بازگشتی writableStream.write(chunk) نشان می‌دهد که آیا بافر داخلی برای داده‌های بیشتر آماده است، که برای مدیریت فشار برگشتی بسیار مهم است:

true : داده ها با موفقیت نوشته شدند و می توانید بلافاصله به نوشتن داده های بیشتری ادامه دهید.

false : بافر داخلی پر است (در حال رسیدن به حد highWaterMark ). این به این معنی نیست که خطایی رخ داده است، بلکه نشان می دهد که باید نوشتن را متوقف کنید تا از بارگذاری بیش از حد بافر جلوگیری کنید. قبل از از سرگیری نوشتن باید منتظر رویداد 'drain' باشید.

یک رویکرد بهتر که فشار برگشتی را کنترل می کند:

 const fs = require ( 'fs' ) ; const readableStream = fs . createReadStream ( 'file1.txt' ) ; const writableStream = fs . createWriteStream ( 'file2.txt' ) ;
readableStream . setEncoding ( 'utf8' ) ;
readableStream . on ( 'data' , function ( chunk ) {  
const canContinue = writableStream . write ( chunk ) ; if ( ! canContinue ) { readableStream . pause ( ) ; } } ) ;
writableStream . on ( 'drain' , function ( ) {  
 readableStream . resume ( ) ; } ) ;
readableStream . on ( 'end' , function ( ) { writableStream . end ( ) ; } ) ;

readableStream . on ( 'error' , ( err ) => { console . error ( 'Read error:' , err ) ; writableStream . end ( ) ; } ) ;
writableStream . on ( 'error' , ( err ) => { console . error ( 'Write error:' , err ) ; } ) ;

این مثال به فشار برگشتی می‌پردازد که یک مفهوم مهم در جریان‌ها است:

هنگامی که write() false را برمی گرداند، به این معنی است که بافر داخلی پر است و ما باید ارسال اطلاعات بیشتر را متوقف کنیم.

ما جریان قابل خواندن را متوقف می کنیم تا دریافت داده ها به طور موقت متوقف شود.

وقتی جریان قابل نوشتن «drain» منتشر می‌کند، به این معنی است که بافر خالی شده است و می‌توانیم خواندن را از سر بگیریم.

ما همچنین مدیریت صحیح خطا را برای هر دو جریان اضافه کرده ایم.

وقتی خواندن کامل شد، end() را در جریان قابل نوشتن برای تکمیل سیگنال فراخوانی می کنیم.

این رویکرد از رشد نامحدود حافظه در زمانی که نویسنده نمی تواند با خواننده همگام شود جلوگیری می کند.

پایان داده ها

هنگامی که اطلاعات بیشتری برای نوشتن ندارید، می توانید به سادگی با end() تماس بگیرید تا به جریان اطلاع دهید که نوشتن را به پایان رسانده اید. با فرض اینکه res یک شی پاسخ HTTP است، معمولاً برای ارسال پاسخ به مرورگر موارد زیر را انجام می دهید:

 res . write ( 'Some Data!!' ) ; res . end ( 'Ended.' ) ;

هنگامی که end() فراخوانی می شود و هر تکه داده پاک می شود، یک رویداد finish توسط جریان منتشر می شود. فقط توجه داشته باشید که پس از فراخوانی end() نمی توانید در جریان بنویسید. به عنوان مثال، موارد زیر منجر به خطا می شود.

 res . write ( 'Some Data!!' ) ; res . end ( ) ; res . write ( 'Trying to write again' ) ;

در اینجا چند events مهم مربوط به جریان های قابل نوشتن آورده شده است:

error – برای نشان دادن اینکه خطا در حین نوشتن/پایپینگ رخ داده است، منتشر می شود.

pipe - وقتی یک جریان قابل خواندن به یک جریان قابل نوشتن لوله می شود، این رویداد توسط جریان قابل نوشتن منتشر می شود.

unpipe - زمانی که unpipe را در جریان قابل خواندن فراخوانی می‌کنید و از لوله‌کشی آن به جریان مقصد جلوگیری می‌کنید، منتشر می‌شود.

جریان های دوبلکس و تبدیل

جریان‌های دوبلکس، جریان‌های خواندنی و قابل نوشتن ترکیبی هستند. آنها دو بافر داخلی جداگانه دارند، یکی برای خواندن و دیگری برای نوشتن، که مستقل از یکدیگر عمل می کنند.

جریان‌های دوبلکس زمانی مفید هستند که به جریان‌های ورودی و خروجی همزمان اما مجزا نیاز دارید، مانند سوکت‌های شبکه (مانند TCP).

 const { Duplex } = require ( 'stream' ) ;

const myDuplex = new Duplex ( { read ( size ) {    
this . push ( String . fromCharCode ( this . currentCharCode ++ ) ) ; if ( this . currentCharCode > 90 ) { this . push ( null ) ; } } , write ( chunk , encoding , callback ) {    
console . log ( chunk . toString ( ) ) ; callback ( ) ; } } ) ;

myDuplex . currentCharCode = 65 ;

این مثال یک جریان Duplex سفارشی ایجاد می کند:

متد read() حروف بزرگ را از A تا Z تولید می کند (کدهای ASCII 65-90).

هر بار که read() فراخوانی می شود، حرف بعدی را فشار می دهد و شمارنده را افزایش می دهد.

وقتی به "Z" رسیدیم، null را فشار می دهیم تا پایان جریان خواندن را نشان دهیم.

متد write() به سادگی تمامی داده های نوشته شده در جریان را در کنسول ثبت می کند.

جریان‌های دوبلکس زمانی مفید هستند که به عملیات خواندن و نوشتن مستقل در یک جریان نیاز دارید.

جریان‌های تبدیل نوع خاصی از جریان‌های دوبلکس هستند که می‌توانند داده‌ها را هنگام نوشتن و خواندن تغییر داده یا تغییر دهند. برخلاف جریان‌های دوبلکس، که ورودی و خروجی از هم جدا هستند، جریان‌های Transform خروجی خود را مستقیماً با ورودی مرتبط می‌کنند. نمونه‌های معمولی عبارتند از جریان‌های zlib برای فشرده‌سازی/فشرده‌سازی و جریان‌های رمزنگاری برای رمزگذاری/رمزگشایی.

 const { Transform } = require ( 'stream' ) ;

const upperCaseTr = new Transform ( { transform ( chunk , encoding , callback ) {    
this . push ( chunk . toString ( ) . toUpperCase ( ) ) ; callback ( ) ; } } ) ;

process . stdin . pipe ( upperCaseTr ) . pipe ( process . stdout ) ;

این مثال جریان تبدیل:

یک جریان تبدیل ایجاد می کند که متن ورودی را به حروف بزرگ تبدیل می کند.

متد transform() تکه های ورودی را می گیرد، آنها را تبدیل می کند و آنها را به خروجی هل می دهد.

ما از ورودی استاندارد، از طریق ترانسفورماتور خود، به خروجی استاندارد لوله کشی می کنیم.

وقتی این کد را اجرا می کنید، هر چیزی که تایپ کنید با حروف بزرگ نمایش داده می شود.

جریان‌های تبدیل برای پردازش یا اصلاح داده‌ها در حین جریان، مانند تجزیه JSON، تبدیل کدگذاری‌ها یا رمزگذاری داده‌ها، ایده‌آل هستند.

نتیجه گیری

این همه در مورد اصول اولیه استریم ها بود. جریان‌ها، لوله‌ها و زنجیره‌بندی اصلی‌ترین و قدرتمندترین ویژگی‌های Node.js هستند. در صورت استفاده مسئولانه، استریم‌ها واقعاً می‌توانند به شما کمک کنند تا کدی منظم و کاربردی برای اجرای I/O بنویسید. فقط مطمئن شوید که خطاهای استریم را مدیریت کرده و استریم ها را به طور مناسب ببندید تا از نشت حافظه جلوگیری شود.

با افزوده‌های جدیدتر به Node.js API مانند stream.pipeline()، stream.finished() و API‌های جریان مبتنی بر Promise، مدیریت جریان‌ها قوی‌تر و کار کردن با آن‌ها آسان‌تر شده است. هنگامی که با مقادیر زیادی داده سروکار دارید، جریان ها باید راه حل شما برای استفاده کارآمد از حافظه و عملکرد باشند.

Node.js Streams چیست؟

خبرکاو