สตรีมใน Node.js: FileStream, Pipes และ Events

ประเภทของสตรีมใน Node.js

สตรีมทุกประเภทที่ Node.js เปิดเผยนั้นจัดอยู่ในหนึ่งในสี่ประเภทพื้นฐาน และการรู้ว่า API ใดส่งคืนประเภทใดจะช่วยอธิบายพฤติกรรมของสตรีมนั้นก่อนที่จะเขียนโค้ดใดๆ โมดูลสตรีมที่มาพร้อมกับ Node.js กำหนดหมวดหมู่เหล่านี้ และ API ไฟล์ เครือข่าย และการบีบอัดที่ใช้ในบทความนี้ทั้งหมดสร้างขึ้นบนพื้นฐานของหมวดหมู่เหล่านี้

สตรีมที่อ่านได้และสตรีมที่เขียนได้เป็นองค์ประกอบพื้นฐานสองอย่างที่ใช้ตลอดบทความนี้ ทั้งสองอย่างนี้ยังเป็นพลังงานขับเคลื่อนอีกด้วย เว็บเซิร์ฟเวอร์ HTTP ใน Node.js สตรีมแบบ Duplex จะรวมบทบาททั้งสองไว้ในอ็อบเจ็กต์เดียว ในขณะที่สตรีมแบบ Transform คือสตรีมแบบ Duplex ที่เอาต์พุตจะปรับเปลี่ยนอินพุต เช่น การบีบอัดแบบ gzip ที่ใช้ระหว่างสตรีม

สตรีมไฟล์ใน Node.js

เมื่อกำหนดประเภทสตรีมทั้งสี่แล้ว ส่วนที่เหลือของบทความนี้จะสร้างการถ่ายโอนไฟล์โดยใช้ API สตรีมที่อ่านได้และเขียนได้

โหนดใช้สตรีมเป็นกลไกการถ่ายโอนข้อมูลอย่างกว้างขวาง

ตัวอย่างเช่น เมื่อคุณส่งออกสิ่งใดไปยังคอนโซลโดยใช้ฟังก์ชัน console.log แสดงว่าคุณกำลังใช้สตรีมเพื่อส่งข้อมูลไปยังคอนโซลจริงๆ

Node.js ยังมีความสามารถในการสตรีมข้อมูลจากไฟล์เพื่อให้สามารถอ่านและเขียนได้อย่างเหมาะสม ตอนนี้เราจะดูตัวอย่างว่าเราสามารถใช้สตรีมเพื่ออ่านและเขียนจากไฟล์ได้อย่างไร เราต้องทำตามขั้นตอนที่กล่าวถึงด้านล่างสำหรับตัวอย่างนี้

ขั้นตอน 1) สร้างไฟล์ชื่อ data.txt ซึ่งมีข้อมูลด้านล่าง สมมติว่าไฟล์นี้ถูกเก็บไว้ในไดรฟ์ D ของเครื่องของเรา

บทช่วยสอนเกี่ยวกับ Node.js

บทนำ

อีเวนต์

เครื่องกำเนิดไฟฟ้า

การเชื่อมต่อข้อมูล

การใช้จัสมิน

ขั้นตอน 2) เขียนโค้ดที่เกี่ยวข้องซึ่งจะใช้ประโยชน์จากสตรีมเพื่ออ่านข้อมูลจากไฟล์

var fs = require("fs");
var stream;
stream = fs.createReadStream("D://data.txt");

stream.on("data", function(data) {
    var chunk = data.toString();
    console.log(chunk);
}); 

Code คำอธิบาย:-

1. ก่อนอื่นเราต้องรวมโมดูล 'fs' ซึ่งมีฟังก์ชันทั้งหมดที่จำเป็นในการสร้างสตรีม
2. ขั้นต่อไป เราจะสร้างสตรีมที่อ่านได้โดยใช้เมธอด createReadStream โดยให้ระบุตำแหน่งของไฟล์ data.txt เป็นอินพุต
3. ฟังก์ชัน steam.on เป็นตัวจัดการเหตุการณ์ และในฟังก์ชันนั้น เรากำลังระบุพารามิเตอร์แรกเป็น 'data' ซึ่งหมายความว่าเมื่อใดก็ตามที่ข้อมูลเข้ามาในสตรีมจากไฟล์ ให้เรียกใช้ฟังก์ชันการโทรกลับ ในกรณีของเรา เรากำลังกำหนดฟังก์ชันการโทรกลับซึ่งจะดำเนินการ 2 ขั้นตอนพื้นฐาน ประการแรกคือการแปลงข้อมูลที่อ่านจากไฟล์เป็นสตริง อย่างที่สองคือการส่งสตริงที่แปลงแล้วเป็นเอาต์พุตไปยังคอนโซล
4. เรากำลังนำข้อมูลแต่ละก้อนที่อ่านจากสตรีมข้อมูลมาแปลงเป็นสตริง
5. ในที่สุด เรากำลังส่งเอาต์พุตของแต่ละสตริงที่แปลงก้อนไปยังคอนโซล

Output:

• หากโค้ดทำงานอย่างถูกต้อง คุณจะเห็นผลลัพธ์ข้างต้นในคอนโซล ผลลัพธ์นี้จะเหมือนกับผลลัพธ์ในไฟล์ data.txt

กำลังเขียนลงในไฟล์

ในทำนองเดียวกัน ที่เราสร้างสตรีมการอ่าน เรายังสามารถสร้างสตรีมการเขียนเพื่อเขียนข้อมูลลงในไฟล์ได้ด้วย ขั้นแรกเรามาสร้างไฟล์เปล่าที่ไม่มีเนื้อหาเรียกว่า data.txt สมมติว่าไฟล์นี้อยู่ในไดรฟ์ D ของคอมพิวเตอร์ของเรา

โค้ดด้านล่างแสดงวิธีที่เราสามารถเขียนข้อมูลลงไฟล์ได้

var fs = require("fs");
var stream;
stream = fs.createWriteStream("D://data.txt");

stream.write("Tutorial on Node.js")
stream.write("Introduction")
stream.write("Events")
stream.write("Generators")
stream.write("Data Connectivity")
stream.write("Using Jasmine") 

Code คำอธิบาย:-

1. เรากำลังสร้างสตรีมแบบเขียนได้โดยใช้วิธี - createWriteStream สำหรับอินพุต เราจะระบุตำแหน่งของไฟล์ data.txt ของเรา
2. ต่อไปเราใช้วิธี stream.write เพื่อเขียนบรรทัดข้อความต่างๆ ลงในไฟล์ข้อความของเรา สตรีมจะดูแลการเขียนข้อมูลนี้ลงในไฟล์ data.txt

หากคุณเปิดไฟล์ data.txt ตอนนี้คุณจะเห็นข้อมูลต่อไปนี้ในไฟล์

บทช่วยสอนเกี่ยวกับ Node.js

บทนำ

อีเวนต์

เครื่องกำเนิดไฟฟ้า

การเชื่อมต่อข้อมูล

การใช้ ดอกมะลิ

ไปป์ใน Node.js

การอ่านและเขียนสตรีมแยกจากกันนั้นมีประโยชน์ แต่ Node.js ยังอนุญาตให้สตรีมที่อ่านได้เชื่อมต่อโดยตรงกับสตรีมที่เขียนได้ ทำให้สามารถย้ายข้อมูลได้โดยไม่ต้องเขียนโค้ดบัฟเฟอร์ด้วยตนเอง

ภายในแอปพลิเคชัน Node สตรีมสามารถไปป์ร่วมกันได้โดยใช้เมธอดไปป์ () ซึ่งรับอาร์กิวเมนต์สองตัว:

• กระแสข้อมูลที่สามารถเขียนได้ที่จำเป็นซึ่งทำหน้าที่เป็นปลายทางสำหรับข้อมูลและ
• ออบเจ็กต์ทางเลือกที่ใช้ในการส่งผ่านตัวเลือก

ตัวอย่างทั่วไปของการใช้ไปป์ หากคุณต้องการถ่ายโอนข้อมูลจากไฟล์หนึ่งไปยังอีกไฟล์หนึ่ง

มาดูตัวอย่างวิธีที่เราสามารถถ่ายโอนข้อมูลจากไฟล์หนึ่งไปยังอีกไฟล์หนึ่งโดยใช้ไพพ์

ขั้นตอน 1) สร้างไฟล์ชื่อ datainput.txt ซึ่งมีข้อมูลด้านล่าง สมมติว่าไฟล์นี้ถูกเก็บไว้ในไดรฟ์ D ของเครื่องของเรา

บทช่วยสอนเกี่ยวกับ Node.js

บทนำ

อีเวนต์

เครื่องกำเนิดไฟฟ้า

การเชื่อมต่อข้อมูล

การใช้จัสมิน

ขั้นตอน 2) สร้างไฟล์เปล่าเปล่าชื่อ dataOutput.txt และวางไว้บนไดรฟ์ D ของเครื่องของคุณ

ขั้นตอน 3) เขียนโค้ดด้านล่างเพื่อดำเนินการถ่ายโอนข้อมูลจากไฟล์ datainput.txt ไปยังไฟล์ dataOutput.txt

var fs = require("fs");
var readStream = fs.createReadStream("D://datainput.txt");
var writeStream = fs.createWriteStream("D://dataOutput.txt");
readStream.pipe(writeStream);

Code คำอธิบาย:-

1. ขั้นแรก เรากำลังสร้าง "readstream" ไปยังไฟล์ datainput.txt ซึ่งมีข้อมูลทั้งหมดของเราซึ่งจำเป็นต้องถ่ายโอนไปยังไฟล์ใหม่
2. จากนั้นเราจำเป็นต้องสร้าง "writestream" ลงในไฟล์ dataOutput.txt ซึ่งเป็นไฟล์เปล่าของเราและเป็นปลายทางสำหรับการถ่ายโอนข้อมูลจากไฟล์ datainput.txt
3. จากนั้นเราใช้คำสั่งไปป์เพื่อถ่ายโอนข้อมูลจาก readstream ไปยังสตรีมการเขียน คำสั่งไปป์จะนำข้อมูลทั้งหมดที่เข้ามาใน readstream และพุชไปที่ writestream

หากตอนนี้คุณเปิดไฟล์ dataOutput.txt คุณจะเห็นข้อมูลทั้งหมดที่มีอยู่ในไฟล์ datainput.txt

Stream Events และ Backpressure ใน Node.js

การเรียกใช้ pipe() ดังที่แสดงข้างต้น จะซ่อนกลไกสองอย่างที่มีความสำคัญในระดับใหญ่ ได้แก่ เหตุการณ์ที่แต่ละสตรีมปล่อยออกมา และระบบแรงดันย้อนกลับที่ป้องกันไม่ให้แหล่งที่มาที่เร็วเกินไปครอบงำปลายทางที่ช้ากว่า

• ข้อมูล: ส่งออกมาเป็นสตรีมที่อ่านได้ทุกครั้งที่มีส่วนข้อมูลใหม่พร้อมใช้งาน
• จบ: ปล่อยออกมาเมื่อสตรีมที่อ่านได้ส่งส่วนสุดท้ายเสร็จแล้ว
• เสร็จสิ้น: ปล่อยออกมาจากสตรีมที่เขียนได้เมื่อข้อมูลที่อยู่ในคิวทั้งหมดถูกส่งออกไปแล้ว
• ข้อผิดพลาด: ข้อผิดพลาดที่เกิดขึ้นจากสตรีมทั้งสองประเภท จำเป็นต้องจัดการกับข้อผิดพลาดนี้เสมอ มิเช่นนั้นกระบวนการจะหยุดทำงาน

ตัวอย่างด้านล่างนี้จะรับฟังเหตุการณ์ทั้งสี่เหตุการณ์พร้อมกับจัดการแรงดันย้อนกลับด้วยตนเอง ซึ่งเป็นพฤติกรรมเดียวกับที่ฟังก์ชัน pipe() ทำโดยอัตโนมัติ

var fs = require("fs");
var readStream = fs.createReadStream("D://data.txt");
var writeStream = fs.createWriteStream("D://dataOutput.txt");

readStream.on("data", function(chunk) {
    var ok = writeStream.write(chunk);
    if (!ok) {
        readStream.pause();
    }
});

writeStream.on("drain", function() {
    readStream.resume();
});

readStream.on("end", function() {
    writeStream.end();
});

writeStream.on("finish", function() {
    console.log("Write completed.");
});

readStream.on("error", function(err) {
    console.log(err);
});

แรงดันย้อนกลับเกิดขึ้นเมื่อสตรีมที่อ่านได้ผลิตข้อมูลเร็วกว่าที่สตรีมที่เขียนได้จะรับได้ เมธอด write() จะส่งคืนค่า false เมื่อบัฟเฟอร์เต็ม ดังนั้นโค้ดด้านบนจะหยุดการทำงานของแหล่งที่มาชั่วคราว จนกว่าปลายทางจะส่งเหตุการณ์ drain ออกมา ซึ่งเป็นการส่งสัญญาณว่าปลอดภัยที่จะดำเนินการต่อ เมธอด pipe() จะดำเนินการวงจรนี้โดยอัตโนมัติ

⚠ คำแนะนำ: ในการใช้งานจริง ควรเลือกใช้ฟังก์ชัน pipeline() จากโมดูล stream แทนการใช้ pipe() ด้วยตนเอง เนื่องจากฟังก์ชัน pipeline() จะส่งต่อข้อผิดพลาดจากทุกสตรีมและปิดไฟล์โดยอัตโนมัติ

เหตุการณ์ใน Node.js

สตรีมเป็นตัวอย่างหนึ่งของการออกแบบที่ขับเคลื่อนด้วยเหตุการณ์ในวงกว้างของ Node.js บทความนี้จะกล่าวถึงวิธีการทำงานของเหตุการณ์โดยทั่วไป โดยใช้คลาส EventEmitter เดียวกันกับที่สตรีมใช้ภายใน

เหตุการณ์เป็นหนึ่งในแนวคิดหลักของ Node.js และบางครั้ง Node.js จะถูกเรียกว่าเฟรมเวิร์กที่ขับเคลื่อนโดยเหตุการณ์

โดยพื้นฐานแล้วเหตุการณ์คือสิ่งที่เกิดขึ้น ตัวอย่างเช่น หากมีการสร้างการเชื่อมต่อกับฐานข้อมูล เหตุการณ์การเชื่อมต่อฐานข้อมูลจะถูกทริกเกอร์ การเขียนโปรแกรมที่ขับเคลื่อนด้วยเหตุการณ์คือการสร้างฟังก์ชันที่จะถูกทริกเกอร์เมื่อมีการทริกเกอร์เหตุการณ์เฉพาะ

มาดูตัวอย่างพื้นฐานของการกำหนดเหตุการณ์ใน Node.js

เราจะสร้างกิจกรรมที่เรียกว่า 'data_received' เมื่อเหตุการณ์นี้เกิดขึ้น ข้อความ "ข้อมูลที่ได้รับ" จะถูกส่งไปยังคอนโซล

var events = require('events');
var eventEmitter = new events.EventEmitter();
eventEmitter.on('data_received', function() {
    console.log('data received succesfully.');
});

eventEmitter.emit('data_received'); 

Code คำอธิบาย:-

1. ใช้ฟังก์ชัน need เพื่อรวมโมดูล 'เหตุการณ์' ด้วยโมดูลนี้ คุณจะสามารถสร้างกิจกรรมใน Node.js ได้
2. สร้างตัวปล่อยเหตุการณ์ใหม่ ใช้เพื่อผูกเหตุการณ์ ซึ่งในกรณีของเราคือ “data_received” ไปยังฟังก์ชันการโทรกลับที่กำหนดไว้ในขั้นตอนที่ 3
3. เรากำหนดฟังก์ชันที่ขับเคลื่อนด้วยเหตุการณ์ซึ่งระบุว่าหากในกรณีที่เหตุการณ์ "data_received" ถูกทริกเกอร์ เราควรส่งออกข้อความ "data_received" ไปยังคอนโซล
4. สุดท้ายนี้ เรามีทริกเกอร์เหตุการณ์ของเราด้วยตนเองโดยใช้ฟังก์ชัน eventEmiter.emit สิ่งนี้จะทริกเกอร์เหตุการณ์ data_received

เมื่อรันโปรแกรม ข้อความ “ข้อมูลที่ได้รับ” จะถูกส่งไปยังคอนโซลดังที่แสดงด้านล่าง

เหตุการณ์ที่เปล่งออกมา

นอกเหนือจากการส่งและรับฟังเหตุการณ์แล้ว Node.js ยังมีเมธอดเพิ่มเติมสำหรับการควบคุมวิธีการลงทะเบียนและตรวจสอบผู้รับฟังอีกด้วย

เมื่อกำหนดเหตุการณ์ มีวิธีการต่างๆ สำหรับเหตุการณ์ที่สามารถเรียกใช้ได้ หัวข้อนี้เน้นไปที่การดูรายละเอียดแต่ละรายการ

1. ตัวจัดการเหตุการณ์ครั้งเดียว

บางครั้งคุณอาจสนใจที่จะตอบสนองต่อเหตุการณ์เฉพาะครั้งแรกที่เกิดขึ้นเท่านั้น ในสถานการณ์เหล่านี้ คุณสามารถใช้เมธอด Once() ได้

มาดูกันว่าเราจะใช้เมธอดครั้งเดียวกับตัวจัดการเหตุการณ์ได้อย่างไร

Code คำอธิบาย:-

1. ที่นี่เราใช้เมธอด 'ครั้งเดียว' เพื่อบอกว่าสำหรับเหตุการณ์ 'data_received' ฟังก์ชันการโทรกลับควรจะดำเนินการเพียงครั้งเดียว
2. ที่นี่เรากำลังทริกเกอร์เหตุการณ์ 'data_received' ด้วยตนเอง
3. เมื่อเหตุการณ์ 'data_received' ถูกทริกเกอร์อีกครั้ง คราวนี้จะไม่มีอะไรเกิดขึ้น นี่เป็นเพราะขั้นตอนแรกที่เราบอกว่าเหตุการณ์จะเกิดขึ้นได้เพียงครั้งเดียวเท่านั้น

หากโค้ดดำเนินการอย่างถูกต้อง ผลลัพธ์ในบันทึกจะเป็น 'data_received สำเร็จ' ข้อความนี้จะปรากฏเพียงครั้งเดียวในคอนโซล

2. การตรวจสอบผู้ฟังเหตุการณ์

ณ จุดใดก็ตามของอายุการใช้งาน ตัวปล่อยเหตุการณ์สามารถมี Listener แนบอยู่เป็นศูนย์หรือมากกว่านั้นได้ Listener สำหรับเหตุการณ์แต่ละประเภทสามารถตรวจสอบได้หลายวิธี

หากคุณสนใจที่จะกำหนดจำนวน Listener ที่แนบมาเท่านั้น ไม่ต้องมองหาที่ไหนไกลไปกว่าเมธอด EventEmitter.listenerCount()

(หมายเหตุ ผู้ฟังมีความสำคัญเนื่องจากโปรแกรมหลักควรทราบว่ามีการเพิ่มผู้ฟังในเหตุการณ์ทันที ไม่เช่นนั้นโปรแกรมจะทำงานผิดพลาดเนื่องจากจะมีการเรียกผู้ฟังเพิ่มเติม)

Code คำอธิบาย:-

1. เรากำลังกำหนดประเภท eventEmitter ซึ่งจำเป็นสำหรับการใช้วิธีการที่เกี่ยวข้องกับเหตุการณ์
2. จากนั้นเรากำลังกำหนดวัตถุที่เรียกว่าตัวปล่อยซึ่งจะใช้ในการกำหนดตัวจัดการเหตุการณ์ของเรา
3. เรากำลังสร้างตัวจัดการเหตุการณ์ 2 ตัวซึ่งโดยพื้นฐานแล้วไม่ได้ทำอะไรเลย ตัวอย่างของเราเก็บตัวอย่างนี้ไว้ง่ายๆ เพื่อแสดงให้เห็นว่าเมธอด ListenerCount ทำงานอย่างไร
4. ตอนนี้เมื่อคุณเรียกใช้เมธอด ListenerCount ในเหตุการณ์ data_received ของเรา ระบบจะส่งจำนวนผู้ฟังเหตุการณ์ที่แนบมากับเหตุการณ์นี้ในบันทึกของคอนโซล

หากโค้ดถูกดำเนินการอย่างถูกต้อง ค่า 2 จะแสดงในบันทึกของคอนโซล

3. กิจกรรม Listener ใหม่

แต่ละครั้งที่มีการลงทะเบียนตัวจัดการเหตุการณ์ใหม่ ตัวปล่อยเหตุการณ์จะปล่อยเหตุการณ์ newListener เหตุการณ์นี้ใช้เพื่อตรวจหาตัวจัดการเหตุการณ์ใหม่ โดยทั่วไปคุณจะใช้เหตุการณ์ newListener เมื่อคุณต้องการจัดสรรทรัพยากรหรือดำเนินการบางอย่างกับตัวจัดการเหตุการณ์ใหม่แต่ละตัว

var events = require('events');
var eventEmitter = events.EventEmitter;
var emitter = new eventEmitter();
emitter.on("newListener", function(eventName, listener) {
    console.log("Added listener for " + eventName + " events");
});
emitter.on('data_received', function() {});
emitter.on('data_received', function() {}); 

Code คำอธิบาย:-

1. เรากำลังสร้างตัวจัดการเหตุการณ์ใหม่สำหรับเหตุการณ์ 'newListener' ดังนั้นเมื่อใดก็ตามที่มีการลงทะเบียนตัวจัดการเหตุการณ์ใหม่ ข้อความ "เพิ่มตัวรับฟังสำหรับ" + ชื่อเหตุการณ์จะแสดงในคอนโซล
2. ที่นี่เรากำลังเขียนข้อความ "Added Listener for" ไปยังคอนโซล + ชื่อเหตุการณ์สำหรับแต่ละเหตุการณ์ที่ลงทะเบียน
3. เรากำลังกำหนดตัวจัดการเหตุการณ์ 2 ตัวสำหรับกิจกรรม 'data_received' ของเรา

หากโค้ดด้านบนทำงานอย่างถูกต้อง ข้อความด้านล่างจะแสดงในคอนโซล มันแสดงให้เห็นว่าตัวจัดการเหตุการณ์ 'newListener' ถูกทริกเกอร์สองครั้ง

เพิ่มผู้ฟังสำหรับเหตุการณ์ data_received

เพิ่มผู้ฟังสำหรับเหตุการณ์ data_received

คำถามที่พบบ่อย