Cách sử dụng sự kiện thoát của luồng.Writable in Node.js

Question

Trong Node.js tôi đang sử dụng phương thức fs.createWriteStream để nối dữ liệu vào tệp cục bộ. Trong tài liệu Node, chúng đề cập đến sự kiện drain khi sử dụng fs.createWriteStream, nhưng tôi không hiểu nó.

var stream = fs.createWriteStream('fileName.txt'); 
var result = stream.write(data); 

Trong mã ở trên, làm cách nào tôi có thể sử dụng sự kiện thoát? Sự kiện có được sử dụng đúng cách bên dưới không?

var data = 'this is my data'; 
if (!streamExists) { 
    var stream = fs.createWriteStream('fileName.txt'); 
} 

var result = stream.write(data); 
if (!result) { 
    stream.once('drain', function() { 
    stream.write(data); 
    }); 
} 

Answer 1

Hãy tưởng tượng bạn đang kết nối 2 luồng với băng thông rất khác nhau, ví dụ: tải tệp cục bộ lên máy chủ chậm. Luồng tệp (nhanh) sẽ phát ra dữ liệu nhanh hơn luồng lưu trữ (chậm) có thể tiêu thụ dữ liệu đó.

Trong trường hợp này, node.js sẽ giữ dữ liệu trong bộ nhớ cho đến khi luồng chậm có cơ hội xử lý nó. Điều này có thể gặp sự cố nếu tệp quá lớn.

Để tránh điều này, Stream.write trả lại false khi bộ đệm hệ thống cơ bản đầy. Nếu bạn ngừng viết, luồng này sau đó sẽ phát ra sự kiện drain để cho biết rằng bộ đệm hệ thống đã làm trống và nó thích hợp để viết lại.

Bạn có thể sử dụng pause/resume luồng có thể đọc và kiểm soát băng thông của luồng có thể đọc được.

Tốt hơn: bạn có thể sử dụng readable.pipe(writable) sẽ thực hiện việc này cho bạn.

EDIT: Có lỗi trong mã của bạn: bất kể số tiền write trả về, dữ liệu của bạn đã được ghi. Bạn không cần phải thử lại. Trong trường hợp của bạn, bạn đang viết data hai lần.

Something như thế này sẽ làm việc:

var packets = […], 
    current = -1; 

function niceWrite() { 
    current += 1; 

    if (current === packets.length) 
    return stream.end(); 

    var nextPacket = packets[current], 
     canContinue = stream.write(nextPacket); 

    // wait until stream drains to continue 
    if (!canContinue) 
    stream.once('drain', niceWrite); 
    else 
    niceWrite(); 
} 

Answer 2

Sự kiện drain là khi đệm nội bộ một dòng ghi đã bị dọn sạch.

Điều này chỉ có thể xảy ra khi kích thước bộ đệm trong vượt quá thuộc tính highWaterMark, là byte dữ liệu tối đa có thể được lưu trữ bên trong bộ đệm trong của dòng ghi cho đến khi nó dừng đọc từ nguồn dữ liệu.

Nguyên nhân của một cái gì đó như thế này có thể là do các thiết lập liên quan đến việc đọc nguồn dữ liệu từ một luồng nhanh hơn có thể được ghi vào tài nguyên khác. Ví dụ, có hai con suối:

var fs = require('fs'); 

var read = fs.createReadStream('./read'); 
var write = fs.createWriteStream('./write'); 

Bây giờ tưởng tượng rằng các tập tin read là trên SSD và có thể đọc ở 500MB/s và write là trên HDD mà chỉ có thể viết ở 150MB/s. Luồng ghi sẽ không thể theo kịp và sẽ bắt đầu lưu trữ dữ liệu trong bộ đệm bên trong. Khi bộ đệm đã đạt đến số highWaterMark, theo mặc định là 16KB, ghi sẽ bắt đầu quay trở lại false và luồng sẽ xếp hàng nội bộ vào cống. Khi độ dài bộ đệm bên trong là 0, thì sự kiện drain được kích hoạt.

Đây là cách một cống hoạt động:

if (state.length === 0 && state.needDrain) { 
    state.needDrain = false; 
    stream.emit('drain'); 
} 

Và đây là những điều kiện tiên quyết cho một cống mà là một phần của writeOrBuffer chức năng:

var ret = state.length < state.highWaterMark; 
state.needDrain = !ret; 

---

Để xem cách sự kiện drain được sử dụng, lấy ví dụ từ tài liệu Node.js.

function writeOneMillionTimes(writer, data, encoding, callback) { 
    var i = 1000000; 
    write(); 
    function write() { 
    var ok = true; 
    do { 
     i -= 1; 
     if (i === 0) { 
     // last time! 
     writer.write(data, encoding, callback); 
     } else { 
     // see if we should continue, or wait 
     // don't pass the callback, because we're not done yet. 
     ok = writer.write(data, encoding); 
     } 
    } while (i > 0 && ok); 
    if (i > 0) { 
     // had to stop early! 
     // write some more once it drains 
     writer.once('drain', write); 
    } 
    } 
} 

Mục tiêu của hàm là viết 1.000.000 lần cho luồng có thể ghi. Điều gì xảy ra là biến số ok được đặt thành true và vòng lặp chỉ thực thi khi ok là đúng. Đối với mỗi vòng lặp lặp lại, giá trị của ok được đặt thành giá trị stream.write(), sẽ trả về false nếu yêu cầu drain. Nếu ok trở thành sai, thì trình xử lý sự kiện cho drain chờ và đang kích hoạt, tiếp tục lại văn bản.

---

Về mã của bạn, bạn không cần sử dụng sự kiện drain vì bạn chỉ viết một lần ngay sau khi mở luồng. Vì bạn chưa viết gì lên luồng, bộ đệm bên trong trống, và bạn sẽ phải viết ít nhất 16KB theo khối để sự kiện drain kích hoạt. Sự kiện drain là để viết nhiều lần với nhiều dữ liệu hơn cài đặt highWaterMark của luồng có thể ghi của bạn.