Node.js به دلیل توانایی خود در مدیریت کارآمد عملیات I/O مشهور است، و در قلب این قابلیت مفهوم جریان نهفته است. جریان‌ها به شما امکان می‌دهند تا داده‌ها را تکه تکه پردازش کنید، نه اینکه همه چیز را به یکباره در حافظه بارگیری کنید - برای مدیریت فایل‌های بزرگ، درخواست‌های شبکه یا داده‌های بلادرنگ. وقتی استریم‌ها را با تایپ قوی TypeScript جفت می‌کنید، ترکیبی قدرتمند دریافت می‌کنید: عملکرد برابر با ایمنی است.

در این راهنما، ما عمیقاً در جریان‌های Node.js غوطه‌ور می‌شویم، انواع آن‌ها را بررسی می‌کنیم و نمونه‌های عملی را با استفاده از TypeScript مرور می‌کنیم. چه یک مبتدی Node.js باشید و چه از علاقه مندان به TypeScript که به دنبال ارتقاء سطح هستید، این پست شما را پوشش می دهد.

چرا جریان ها اهمیت دارند؟

این را تصور کنید: شما وظیفه دارید یک فایل لاگ 50 گیگابایتی را پردازش کنید. بارگیری کامل آن در حافظه، منابع سرور شما را تمام می کند و منجر به خرابی یا عملکرد کند می شود. جریان‌ها این مشکل را با اجازه دادن به شما در مدیریت داده‌ها در حین جریان حل می‌کنند، مانند جرعه جرعه کشیدن از نی به جای برداشتن یک کوزه گالن.

این کارایی به همین دلیل است که استریم ها سنگ بنای Node.js هستند و همه چیز از عملیات فایل گرفته تا سرورهای HTTP را تامین می کنند. TypeScript با افزودن تعاریف نوع، تشخیص خطاها در زمان کامپایل و بهبود خوانایی کد این امر را تقویت می کند. بیایید به اصول اولیه بپردازیم و ببینیم این هم افزایی در عمل چگونه عمل می کند.

چهار نوع جریان

Node.js چهار نوع جریان اصلی را ارائه می دهد که هر کدام هدف خاصی دارند:

جریان‌های قابل خواندن : منابع داده‌ای که می‌توانید از آنها بخوانید (به عنوان مثال، فایل‌ها، پاسخ‌های HTTP).

جریان‌های قابل نوشتن : مقاصدی که می‌توانید در آنها بنویسید (مثلاً فایل‌ها، درخواست‌های HTTP).

جریان های دوبلکس : هم قابل خواندن و هم قابل نوشتن (به عنوان مثال، سوکت های TCP).

Transform Streams : یک جریان دوبلکس ویژه که داده ها را هنگام عبور تغییر می دهد (مثلاً فشرده سازی).

TypeScript این را با اجازه دادن به ما برای تعریف واسط برای داده های جریان یافته از طریق آنها افزایش می دهد. بیایید آنها را با مثال هایی تجزیه کنیم.

تنظیم محیط TypeScript شما

قبل از اینکه وارد کد شویم، مطمئن شوید که Node.js و TypeScript را نصب کرده اید.

ایجاد یک پروژه جدید:

 mkdir node-streams-typescript cd node-streams-typescript npm init -y npm install typescript @types/node --save-dev npx tsc --init

tsconfig.json خود را به روز کنید تا شامل موارد زیر باشد:

 { "compilerOptions" : { "target" : "ES2020" , "module" : "commonjs" , "strict" : true , "outDir" : "./dist" } , "include" : [ "src/**/*" ] }

یک پوشه src ایجاد کنید و اجازه دهید شروع به کدنویسی کنیم!

مثال 1: خواندن یک فایل با یک جریان قابل خواندن

بیایید یک فایل متنی را تکه تکه بخوانیم. ابتدا، یک فایل به نام data.txt در فهرست اصلی پروژه خود با مقداری متن نمونه ایجاد کنید (به عنوان مثال، "Hello, streams!").

اکنون در src/readStream.ts:

 import { createReadStream } from 'fs' ; import { Readable } from 'stream' ;
const readStream : Readable = createReadStream ( 'data.txt' , { encoding : 'utf8' } ) ;
readStream . on ( 'data' , ( chunk : string ) => { console . log ( 'Chunk received:' , chunk ) ; } ) . on ( 'end' , ( ) => { console . log ( 'Finished reading the file.' ) ; } ) . on ( 'error' , ( err : Error ) => { console . error ( 'Error:' , err . message ) ; } ) ;

اجراش کن با:

 npx tsc && node dist/readStream.js

در اینجا، TypeScript اطمینان حاصل می کند که قطعه به رابط Chunk ما پایبند است، و کنترل کننده رویداد خطا انتظار یک نوع خطا را دارد. این جریان داده ها.txt را به صورت تکه ای می خواند (64 کیلوبایت پیش فرض برای فایل ها) و آنها را ثبت می کند.

مثال 2: نوشتن داده با یک جریان قابل نوشتن

حالا بیایید داده ها را در یک فایل جدید بنویسیم. در src/writeStream.ts:

 import { createWriteStream } from 'fs' ; import { Writable } from 'stream' ;
const writeStream : Writable = createWriteStream ( 'output.txt' , { encoding : 'utf8' } ) ;
const data : string [ ] = [ 'Line 1\n' , 'Line 2\n' , 'Line 3\n' ] ;
data . forEach ( ( line : string ) => { writeStream . write ( line ) ; } ) ;
writeStream . end ( ( ) => { console . log ( 'Finished writing to output.txt' ) ; } ) ;
writeStream . on ( 'error' , ( err : Error ) => { console . error ( 'Error:' , err . message ) ; } ) ;

کامپایل و اجرا کنید:

 npx tsc && node dist/writeStream.js

این output.txt با سه خط ایجاد می کند. TypeScript تضمین می کند که خط یک رشته است و تکمیل خودکار را برای متدهای استریم فراهم می کند.

مثال 3: لوله کشی با جریان تبدیل

Piping جایی است که جریان ها می درخشند و یک جریان قابل خواندن را به یک جریان قابل نوشتن متصل می کنند. بیایید یک چرخش با یک جریان Transform اضافه کنیم تا متن خود را بزرگ کنیم.

در src/transformStream.ts:

 import { createReadStream , createWriteStream } from 'fs' ; import { Transform , TransformCallback } from 'stream' ;

class UppercaseTransform extends Transform { _transform ( chunk : Buffer , encoding : string , callback : TransformCallback ) : void { const upperChunk = chunk . toString ( ) . toUpperCase ( ) ; this . push ( upperChunk ) ; callback ( ) ; } }
const readStream = createReadStream ( 'data.txt' , { encoding : 'utf8' } ) ; const writeStream = createWriteStream ( 'output_upper.txt' ) ; const transformStream = new UppercaseTransform ( ) ;
readStream . pipe ( transformStream ) . pipe ( writeStream ) . on ( 'finish' , ( ) => { console . log ( 'Transform complete! Check output_upper.txt' ) ; } ) . on ( 'error' , ( err : Error ) => { console . error ( 'Error:' , err . message ) ; } ) ;

اجراش کن:

 npx tsc && node dist/transformStream.js

این data.txt را می خواند، متن را به حروف بزرگ تبدیل می کند و آن را در output_upper.txt می نویسد.

نوع TransformCallback TypeScript تضمین می کند که روش _transform ما به درستی پیاده سازی شده است.

مثال 4: فشرده سازی فایل ها با جریان دوبلکس

بیایید با یک سناریوی پیشرفته‌تر مقابله کنیم: فشرده‌سازی یک فایل با استفاده از ماژول zlib، که یک جریان دوطرفه را فراهم می‌کند. همراه با بسته «@types/node» است که قبلاً آن را نصب کرده بودیم.

در src/compressStream.ts:

 import { createReadStream , createWriteStream } from 'fs' ; import { createGzip } from 'zlib' ; import { pipeline } from 'stream' ;
const source = createReadStream ( 'data.txt' ) ; const destination = createWriteStream ( 'data.txt.gz' ) ; const gzip = createGzip ( ) ;
pipeline ( source , gzip , destination , ( err : Error | null ) => { if ( err ) { console . error ( 'Compression failed:' , err . message ) ; return ; } console . log ( 'File compressed successfully! Check data.txt.gz' ) ; } ) ;

اجراش کن:

 npx tsc && node dist/compressStream.js

در اینجا، خط لوله مدیریت صحیح خطا و پاکسازی را تضمین می کند. جریان gzip data.txt را به data.txt.gz فشرده می کند. استنتاج نوع TypeScript کد ما را تمیز و ایمن نگه می دارد.

مثال 5: پخش جریانی پاسخ های HTTP

جریان ها در عملیات شبکه می درخشند. بیایید جریان داده ها را از یک سرور HTTP با استفاده از axios شبیه سازی کنیم. نصبش کن:

 npm install axios @types/axios

در src/httpStream.ts:

 import axios from 'axios' ; import { createWriteStream } from 'fs' ; import { Writable } from 'stream' ;
async function streamHttpResponse ( url : string , outputFile : string ) : Promise < void > { const response = await axios ( { method : 'get' , url , responseType : 'stream' , } ) ;
const writeStream : Writable = createWriteStream ( outputFile ) ; response . data . pipe ( writeStream ) ;
return new Promise ( ( resolve , reject ) => { writeStream . on ( 'finish' , ( ) => { console . log ( ` Downloaded to ${ outputFile } ` ) ; resolve ( ) ; } ) ; writeStream . on ( 'error' , ( err : Error ) => { console . error ( 'Download failed:' , err . message ) ; reject ( err ) ; } ) ; } ) ; }
streamHttpResponse ( 'https://example.com' , 'example.html' ) . catch ( console . error ) ;

اجراش کن:

 npx tsc && node dist/httpStream.js

این یک پاسخ HTTP (به عنوان مثال، یک صفحه وب) را به example.html پخش می کند. TypeScript تضمین می کند که پارامترهای url و outputFile رشته ای هستند و تایپ Promise وضوح را اضافه می کند.

همچنین می‌توانیم از Fetch API داخلی Node.js (در دسترس از Node v18) یا کتابخانه‌هایی مانند node-fetch استفاده کنیم که از پاسخ‌های جریانی نیز پشتیبانی می‌کنند، اگرچه انواع جریان ممکن است متفاوت باشد (Web Streams در مقابل Node.js Streams).

مثال:

 const response = await fetch ( 'https://example.com' ) ; const writeStream = createWriteStream ( outputFile ) ; response . body . pipe ( writeStream ) ;

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

بیایید یک جریان سفارشی و قابل خواندن برای شبیه‌سازی داده‌های بلادرنگ، مانند قرائت‌های حسگر، ایجاد کنیم. در src/customReadable.ts:

 import { Readable } from 'stream' ;
class SensorStream extends Readable { private count : number = 0 ; private max : number = 10 ;
constructor ( options ? : any ) { super ( options ) ; }
_read ( ) : void { if ( this . count < this . max ) { const data = ` Sensor reading ${ this . count } : ${ Math . random ( ) * 100 } \n ` ; this . push ( data ) ; this . count ++ ; } else { this . push ( null ) ; } } }
const sensor = new SensorStream ( { encoding : 'utf8' } ) ;
sensor . on ( 'data' , ( chunk : string ) => { console . log ( 'Received:' , chunk . trim ( ) ) ; } ) . on ( 'end' , ( ) => { console . log ( 'Sensor stream complete.' ) ; } ) . on ( 'error' , ( err : Error ) => { console . error ( 'Error:' , err . message ) ; } ) ;

اجراش کن:

 npx tsc && node dist/customReadable.js

این 10 "خوانش حسگر" تصادفی ایجاد می کند و آنها را پخش می کند. تایپ کلاس TypeScript تضمین می کند که پیاده سازی ما با رابط Readable هماهنگ است.

مثال 7: زنجیر کردن جریان های تبدیل چندگانه

بیایید تبدیل‌ها را زنجیره‌ای کنیم تا متن را در مراحل پردازش کنیم: آن را با حروف بزرگ بنویسیم، سپس یک مهر زمانی اضافه کنیم. در src/chainTransform.ts:

 import { createReadStream , createWriteStream } from 'fs' ; import { Transform , TransformCallback } from 'stream' ;
class UppercaseTransform extends Transform { _transform ( chunk : Buffer , encoding : string , callback : TransformCallback ) : void { this . push ( chunk . toString ( ) . toUpperCase ( ) ) ; callback ( ) ; } }
class TimestampTransform extends Transform { _transform ( chunk : Buffer , encoding : string , callback : TransformCallback ) : void { const timestamp = new Date ( ) . toISOString ( ) ; this . push ( ` [ ${ timestamp } ] ${ chunk . toString ( ) } ` ) ; callback ( ) ; } }
const readStream = createReadStream ( 'data.txt' , { encoding : 'utf8' } ) ; const writeStream = createWriteStream ( 'output_chain.txt' ) ; const upper = new UppercaseTransform ( ) ; const timestamp = new TimestampTransform ( ) ;
readStream . pipe ( upper ) . pipe ( timestamp ) . pipe ( writeStream ) . on ( 'finish' , ( ) => { console . log ( 'Chained transform complete! Check output_chain.txt' ) ; } ) . on ( 'error' , ( err : Error ) => { console . error ( 'Error:' , err . message ) ; } ) ;

اجراش کن:

 npx tsc && node dist/chainTransform.js

این data.txt را می‌خواند، داده‌ها را بزرگ می‌کند، مهر زمانی اضافه می‌کند و نتیجه را در output_chain.txt می‌نویسد. تبدیل‌های زنجیره‌ای، ماژولار بودن جریان‌ها را نشان می‌دهد.

بهترین روش ها برای استریم ها در TypeScript

داده های خود را تایپ کنید : رابط هایی را برای تکه ها تعریف کنید تا زودتر خطاهای نوع را پیدا کنید.

کنترل خطاها : همیشه شنوندگان رویداد خطا را پیوست کنید تا از استثنائات کنترل نشده جلوگیری کنید.

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

فشار برگشتی : برای داده‌های بزرگ، writeStream.writableHighWaterMark را کنترل کنید تا از تحت فشار قرار دادن مقصد جلوگیری کنید.

مورد استفاده در دنیای واقعی: پاسخ‌های API جریانی

تصور کنید در حال ساختن یک API هستید که مجموعه داده بزرگی را پخش می کند. استفاده از اکسپرس و استریم:

 import express from 'express' ; import { Readable } from 'stream' ;
const app = express ( ) ;
app . get ( '/stream-data' , ( req , res ) => { const data = [ 'Item 1\n' , 'Item 2\n' , 'Item 3\n' ] ; const stream = Readable . from ( data ) ;
 res . setHeader ( 'Content-Type' , 'text/plain' ) ; stream . pipe ( res ) ; } ) ;
app . listen ( 3000 , ( ) => { console . log ( 'Server running on port 3000' ) ; } ) ;

وابستگی ها را نصب کنید (npm install express @types/express)، سپس آن را اجرا کنید. برای مشاهده جریان داده ها در مرورگر خود به http://localhost:3000/stream-data مراجعه کنید!

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

هنگامی که یک جریان قابل نوشتن نمی تواند با یک جریان قابل خواندن مطابقت داشته باشد، فشار برگشتی رخ می دهد. Node.js به طور خودکار با لوله ها این کار را انجام می دهد، اما می توانید آن را به صورت دستی نظارت کنید:

 const writeStream = createWriteStream ( 'large_output.txt' ) ;
if ( ! writeStream . write ( 'data' ) ) { console . log ( 'Backpressure detected! Pausing...' ) ; writeStream . once ( 'drain' , ( ) => { console . log ( 'Resuming...' ) ; } ) ; }

این تضمین می کند که برنامه شما تحت بارهای سنگین پاسخگو باقی می ماند.

اقدامات احتیاطی برای استفاده از Backpressure: هنگام نوشتن مقادیر زیاد داده، جریان قابل خواندن ممکن است سریعتر از جریان قابل نوشتن داده تولید کند. در حالی که لوله و خط لوله به طور خودکار این کار را انجام می دهند، اگر به صورت دستی می نویسید، بررسی کنید که آیا write() false برمی گرداند و قبل از نوشتن موارد بیشتر منتظر رویداد «drain» باشید.

به‌علاوه، تکرارکننده‌های غیرهمگام (برای انتظار... از) جایگزین‌های مدرنی برای مصرف جریان‌های قابل خواندن هستند، که اغلب می‌توانند کد را در مقایسه با استفاده از .on('data') و .on ('end') ساده کنند.

مثال:

 async function processStream ( readable : Readable ) { for await ( const chunk of readable ) { console . log ( 'Chunk:' , chunk ) ; } console . log ( 'Finished reading.' ) ; }

نکات اضافی:

اطمینان از پاکسازی منابع: این امر به ویژه در اجرای جریان سفارشی یا هنگام استفاده از stream.pipeline بسیار مهم است. در سناریوهای خطا یا زمانی که جریان دیگر برای آزاد کردن منابع زیربنایی و جلوگیری از نشت نیازی به جریان نیست، به صراحت stream.destroy() را فراخوانی کنید. stream.pipeline به طور خودکار این کار را برای جریان های لوله شده انجام می دهد.

از Readable.from() برای راحتی استفاده کنید: هنگامی که نیاز به ایجاد یک جریان از یک تکرارپذیر موجود (مانند یک آرایه) یا یک تکرار غیر همگام دارید، Readable.from() اغلب ساده‌ترین و مدرن‌ترین رویکرد است که به کد دیگ بخار کمتری نسبت به ایجاد یک کلاس Readable سفارشی نیاز دارد.

نتیجه گیری

Stream ها یک تغییر دهنده بازی در Node.js هستند و TypeScript با معرفی ایمنی و وضوح نوع آنها را بیشتر تقویت می کند. از خواندن فایل‌ها تا تبدیل داده‌ها در زمان واقعی، تسلط بر جریان‌ها دنیایی از امکانات ورودی/خروجی کارآمد را باز می‌کند. مثال‌های اینجا - خواندن، نوشتن، تغییر، فشرده‌سازی، و پخش جریانی از طریق HTTP- سطح آنچه را که ممکن است نشان می‌دهد.

با خطوط لوله خود آزمایش کنید: گزارش‌های جریانی، پردازش فایل‌های CSV یا ایجاد یک سیستم چت زنده را امتحان کنید. هرچه بیشتر کاوش کنید، بیشتر از تطبیق پذیری جریان ها قدردانی خواهید کرد.

خبرکاو