Pemahaman mendalam tentang 4 jenis strim dalam Node.js

Artikel ini akan membawa anda melalui 4 jenis strim dalam Node.js dan melihat cara menyelesaikan masalah "tekanan belakang" letupan penimbal Rakan yang memerlukan boleh mempelajarinya~

Bagaimana untuk mengalihkan sesuatu dari A ke B?

Angkat sahaja, gerakkannya ke destinasi, dan letakkan ke bawah.

Bagaimana jika benda ini seberat satu tan?

Kemudian gerakkan bahagian demi bahagian.

Sebenarnya, IO bermaksud benda bergerak, termasuk IO rangkaian dan IO fail Jika jumlah data adalah kecil, maka ia cukup untuk memindahkan keseluruhan kandungan, tetapi jika terdapat banyak kandungan, memuatkan. ia ke dalam memori pada satu masa akan ranap, dan Kelajuan juga perlahan, jadi anda boleh memprosesnya bahagian demi bahagian Ini adalah idea aliran. [Pembelajaran yang disyorkan: "tutorial nodejs"]

Pelbagai bahasa pada asasnya telah melaksanakan api strim, dan Node.js juga telah melaksanakan api aliran Strim api secara relatifnya biasa digunakan . Mari terokainya di bawah.

Artikel ini akan menjawab soalan berikut:

• Apakah empat aliran Node.js
• Bagaimanakah penjana bergabung dengan Strim Boleh Dibaca
• strim Jeda dan Aliran
• Apakah masalah tekanan belakang dan cara menyelesaikannya

4 jenis aliran Node.js

Aliran Perasaan intuitif

mengalir dari satu tempat ke tempat yang lain Jelas sekali terdapat sisi aliran keluar dan sisi aliran masuk ialah aliran yang boleh dibaca (boleh dibaca), dan bahagian aliran masuk ialah aliran boleh tulis (boleh ditulis).

Sudah tentu ada juga aliran yang boleh mengalir masuk dan keluar Ini dipanggil aliran dupleks (dupleks)

Memandangkan ia boleh mengalir masuk dan keluar, bolehkah kita menukar kandungan masuk ke bawah dan kemudian mengalirkannya keluar. Aliran ini dipanggil aliran transformasi

aliran dupleks? kandungan aliran masuk dan aliran keluar tidak perlu dikaitkan, tetapi aliran masuk dan aliran keluar adalah berkaitan. Ini adalah perbezaan antara keduanya.

Strim api

Strim yang disediakan oleh Node.js ialah empat jenis yang diperkenalkan di atas:

const stream = require('stream');

// 可读流
const Readable = stream.Readable;
// 可写流
const Writable = stream.Writable;
// 双工流
const Duplex = stream.Duplex;
// 转换流
const Transform = stream.Transform;

Semuanya perlu dilaksanakan Kaedah :

Boleh dibaca perlu melaksanakan kaedah _baca untuk mengembalikan kandungan
• Boleh tulis perlu melaksanakan kaedah _tulis untuk menerima kandungan
• Dupleks perlu melaksanakan _baca dan kaedah _tulis Untuk menerima dan mengembalikan kandungan
• Transform perlu melaksanakan kaedah _transform untuk menukar kandungan yang diterima dan mengembalikan

Mari kita lihat secara berasingan:

Boleh dibaca

Boleh dibaca perlu melaksanakan kaedah _baca dan mengembalikan data tertentu melalui tolak.

const Stream = require('stream');

const readableStream = Stream.Readable();

readableStream._read = function() {
    this.push('阿门阿前一棵葡萄树，');
    this.push('阿东阿东绿的刚发芽，');
    this.push('阿东背着那重重的的壳呀，');
    this.push('一步一步地往上爬。')
    this.push(null);
}

readableStream.on('data', (data)=> {
    console.log(data.toString())
});

readableStream.on('end', () => {
    console.log('done~');
});

Apabila menolak null, ini bermakna tamat strim.

Kesan pelaksanaan adalah seperti berikut:

Mencipta Boleh Dibaca juga boleh dilakukan melalui pewarisan:

const Stream = require('stream');

class ReadableDong extends Stream.Readable {

    constructor() {
        super();
    }

    _read() {
        this.push('阿门阿前一棵葡萄树，');
        this.push('阿东阿东绿的刚发芽，');
        this.push('阿东背着那重重的的壳呀，');
        this.push('一步一步地往上爬。')
        this.push(null);
    }

}

const readableStream = new ReadableDong();

readableStream.on('data', (data)=> {
    console.log(data.toString())
});

readableStream.on('end', () => {
    console.log('done~');
});

Strim boleh dibaca menjana kandungan , maka ia secara semula jadi boleh digabungkan dengan penjana:

const Stream = require('stream');

class ReadableDong extends Stream.Readable {

    constructor(iterator) {
        super();
        this.iterator = iterator;
    }

    _read() {
        const next = this.iterator.next();
        if(next.done) {
            return this.push(null);
        } else {
            this.push(next.value)
        }
    }

}

function *songGenerator() {
    yield '阿门阿前一棵葡萄树，';
    yield '阿东阿东绿的刚发芽，';
    yield '阿东背着那重重的的壳呀，';
    yield '一步一步地往上爬。';
}

const songIterator = songGenerator();

const readableStream = new ReadableDong(songIterator);

readableStream.on('data', (data)=> {
    console.log(data.toString())
});

readableStream.on('end', () => {
    console.log('done~');
});

Ini ialah strim boleh dibaca, yang mengembalikan kandungan dengan melaksanakan kaedah _read.

Boleh Ditulis

Boleh Ditulis melaksanakan kaedah _write untuk menerima kandungan bertulis.

const Stream = require('stream');

const writableStream = Stream.Writable();

writableStream._write = function (data, enc, next) {
   console.log(data.toString());
   // 每秒写一次
   setTimeout(() => {
       next();
   }, 1000);
}

writableStream.on('finish', () => console.log('done~'));

writableStream.write('阿门阿前一棵葡萄树，');
writableStream.write('阿东阿东绿的刚发芽，');
writableStream.write('阿东背着那重重的的壳呀，');
writableStream.write('一步一步地往上爬。');
writableStream.end();

Menerima kandungan bertulis, mencetaknya dan memanggil di sebelah untuk memproses kandungan bertulis seterusnya Panggilan seterusnya di sini adalah tak segerak dan kekerapan boleh dikawal.

Selepas menjalankannya, kandungan bertulis sememangnya boleh diproses seperti biasa:

Ini ialah strim boleh tulis, yang mengendalikan penulisan dengan melaksanakan kandungan kaedah _write.

Dupleks

Dupleks boleh dibaca dan boleh ditulis, cuma laksanakan _baca dan _tulis pada masa yang sama

const Stream = require('stream');

var duplexStream = Stream.Duplex();

duplexStream._read = function () {
    this.push('阿门阿前一棵葡萄树，');
    this.push('阿东阿东绿的刚发芽，');
    this.push('阿东背着那重重的的壳呀，');
    this.push('一步一步地往上爬。')
    this.push(null);
}

duplexStream._write = function (data, enc, next) {
    console.log(data.toString());
    next();
}

duplexStream.on('data', data => console.log(data.toString()));
duplexStream.on('end', data => console.log('read done~'));

duplexStream.write('阿门阿前一棵葡萄树，');
duplexStream.write('阿东阿东绿的刚发芽，');
duplexStream.write('阿东背着那重重的的壳呀，');
duplexStream.write('一步一步地往上爬。');
duplexStream.end();

duplexStream.on('finish', data => console.log('write done~'));

Sepadukan strim Boleh Baca dan Fungsi Strim boleh tulis, ini ialah Dupleks aliran dupleks.

Transformasi

Walaupun aliran Dupleks boleh dibaca dan ditulis, tiada kaitan antara kedua-duanya, dan kadangkala ia perlu Selepas menukar kandungan masuk, ia mengalir keluar Pada masa ini, anda perlu menukar aliran Transform.

Strim Transform perlu melaksanakan _transform api Kami melaksanakan aliran penukaran yang membalikkan kandungan:

const Stream = require('stream');

class TransformReverse extends Stream.Transform {

  constructor() {
    super()
  }

  _transform(buf, enc, next) {
    const res = buf.toString().split('').reverse().join('');
    this.push(res)
    next()
  }
}

var transformStream = new TransformReverse();

transformStream.on('data', data => console.log(data.toString()))
transformStream.on('end', data => console.log('read done~'));

transformStream.write('阿门阿前一棵葡萄树');
transformStream.write('阿东阿东绿的刚发芽');
transformStream.write('阿东背着那重重的的壳呀');
transformStream.write('一步一步地往上爬');
transformStream.end()

transformStream.on('finish', data => console.log('write done~'));

Selepas menjalankannya, kesannya adalah seperti berikut:

流的暂停和流动

我们从 Readable 流中获取内容，然后流入 Writable 流，两边分别做 _read 和 _write 的实现，就实现了流动。

背压

但是 read 和 write 都是异步的，如果两者速率不一致呢？

如果 Readable 读入数据的速率大于 Writable 写入速度的速率，这样就会积累一些数据在缓冲区，如果缓冲的数据过多，就会爆掉，会丢失数据。

而如果  Readable 读入数据的速率小于 Writable 写入速度的速率呢？那没关系，最多就是中间有段空闲时期。

这种读入速率大于写入速率的现象叫做“背压”，或者“负压”。也很好理解，写入段压力比较大，写不进去了，会爆缓冲区，导致数据丢失。

这个缓冲区大小可以通过 readableHighWaterMark 和 writableHightWaterMark 来查看，是 16k。

解决背压

怎么解决这种读写速率不一致的问题呢？

当没写完的时候，暂停读就行了。这样就不会读入的数据越来越多，驻留在缓冲区。

readable stream 有个 readableFlowing 的属性，代表是否自动读入数据，默认为 true，也就是自动读入数据，然后监听 data 事件就可以拿到了。

当 readableFlowing 设置为 false 就不会自动读了，需要手动通过 read 来读入。

readableStream.readableFlowing = false;

let data;
while((data = readableStream.read()) != null) {
    console.log(data.toString());
}

但自己手动 read 比较麻烦，我们依然可以用自动流入的方式，调用 pause 和 resume 来暂停和恢复就行了。

当调用 writable stream 的 write 方法的时候会返回一个 boolean 值代表是写入了目标还是放在了缓冲区：

• true: 数据已经写入目标
• false：目标不可写入，暂时放在缓冲区

我们可以判断返回 false 的时候就 pause，然后等缓冲区清空了就 resume：

const rs = fs.createReadStream(src);
const ws = fs.createWriteStream(dst);

rs.on('data', function (chunk) {
    if (ws.write(chunk) === false) {
        rs.pause();
    }
});

rs.on('end', function () {
    ws.end();
});

ws.on('drain', function () {
    rs.resume();
});

这样就能达到根据写入速率暂停和恢复读入速率的功能，解决了背压问题。

pipe 有背压问题么？

平时我们经常会用 pipe 来直接把 Readable 流对接到 Writable 流，但是好像也没遇到过背压问题，其实是 pipe 内部已经做了读入速率的动态调节了。

const rs = fs.createReadStream(src);
const ws = fs.createWriteStream(dst);

rs.pipe(ws);

总结

流是传输数据时常见的思想，就是一部分一部分的传输内容，是文件读写、网络通信的基础概念。

Node.js 也提供了 stream 的 api，包括 Readable 可读流、Writable 可写流、Duplex 双工流、Transform 转换流。它们分别实现 _read、_write、_read + _write、_transform 方法，来做数据的返回和处理。

创建 Readable 对象既可以直接调用 Readable api 创建，然后重写 _read 方法，也可以继承 Readable 实现一个子类，之后实例化。其他流同理。（Readable 可以很容易的和 generator 结合）

当读入的速率大于写入速率的时候就会出现“背压”现象，会爆缓冲区导致数据丢失，解决的方式是根据 write 的速率来动态 pause 和 resume 可读流的速率。pipe 就没有这个问题，因为内部做了处理。

流是掌握 IO 绕不过去的一个概念，而背压问题也是流很常见的问题，遇到了数据丢失可以考虑是否发生了背压。希望这篇文章能够帮大家理清思路，真正掌握 stream！

更多编程相关知识，请访问：编程入门！！

Atas ialah kandungan terperinci Pemahaman mendalam tentang 4 jenis strim dalam Node.js. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!