Compréhension approfondie des 4 types de flux dans Node.js

Cet article vous amènera à comprendre les 4 types de flux dans Node.js et à voir comment résoudre le problème de « contre-pression » du tampon Les amis dans le besoin peuvent apprendre et comprendre ~

Mettez-en un. Comment déplacer des choses de A à B ?

Il suffit de le soulever, de le déplacer vers la destination et de le poser.

Et si cette chose pesait une tonne ?

Puis déplacez-le partie par partie.

En fait, IO signifie déplacer des éléments, y compris les IO réseau et les IO fichiers. Si la quantité de données est faible, alors tout le contenu peut être transféré directement. Mais s'il y a beaucoup de contenu, chargez-le dans la mémoire en une seule fois. le temps va s'écraser et la vitesse sera lente. À ce moment-là, vous pouvez le traiter partie par partie. C'est l'idée du flux. [Apprentissage recommandé : "Tutoriel Nodejs"]

Divers langages ont essentiellement implémenté l'API de flux, et Node.js a également implémenté l'API de flux. L'API de flux est plus couramment utilisée. Explorons le flux ci-dessous.

Cet article répondra aux questions suivantes :

• Quels sont les 4 flux de Node.js
• Comment le générateur se combine-t-il avec Readable Stream
• La pause et le flux des flux
• Quel est le problème de contre-pression et comment pour le résoudre

Les quatre types de flux dans Node.js

La sensation intuitive des flux

coule d'un endroit à un autre. Évidemment, il y a une sortie et une entrée. La sortie est un flux lisible. (lisible), et l'afflux Un côté est le flux inscriptible (inscriptible).

Bien sûr, il existe également des flux qui peuvent à la fois entrer et sortir. C'est ce qu'on appelle le flux duplex.

Puisqu'il peut entrer et sortir, pouvons-nous convertir le contenu entrant ? , ce type de flux est appelé flux de transformation (transformation)

flux duplex. Le contenu entrant et sortant du flux duplex n'a pas besoin d'être lié, tandis que le flux entrant et sortant du flux de transformation est lié. est la différence entre les deux.

Stream api

Les flux fournis par Node.js sont les 4 types présentés ci-dessus :

const stream = require('stream');

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

Ils ont tous des méthodes à implémenter :

• Readable doit implémenter la méthode _read pour renvoyer du contenu
• Besoins en écriture Implémentez la méthode _write pour accepter le contenu
• Duplex doit implémenter les méthodes _read et _write pour accepter et renvoyer le contenu
• Transform doit implémenter la méthode _transform pour convertir le contenu reçu et le renvoyer

Jetons un coup d'œil à chacun :

Readable

Readable doit implémenter la méthode _read et renvoyer des données spécifiques via push.

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~');
});

Lorsque vous poussez un null, cela signifie la fin du flux.

L'effet d'exécution est le suivant :

La création de Readable peut également se faire par héritage :

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~');
});

Le flux lisible génère du contenu, il peut donc naturellement être combiné avec le générateur :

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~');
});

C'est le flux lisible , via Implémentez la méthode _read pour renvoyer le contenu.

Writable

Writable Pour implémenter la méthode _write, recevez le contenu écrit.

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();

Reçoit le contenu écrit, l'imprime et appelle ensuite pour traiter le prochain contenu écrit. Le prochain appel ici est asynchrone et la fréquence peut être contrôlée.

Après une exécution pendant un certain temps, le contenu écrit peut en effet être traité normalement :

Il s'agit d'un flux inscriptible, et le contenu écrit est traité en implémentant la méthode _write.

Duplex

Duplex est lisible et inscriptible, il suffit d'implémenter _read et _write en même temps

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~'));

Intégrez les fonctions de flux lisible et de flux inscriptible, c'est le flux duplex Duplex.

Transform

Bien que le flux Duplex puisse être lu et écrit, il n'y a aucune connexion entre les deux. Parfois, le contenu entrant doit être converti puis diffusé. Dans ce cas, le flux Transform doit le faire. être transformé.

Transform stream doit implémenter l'API _transform Nous implémentons le flux de transformation qui inverse le contenu :

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~'));

Exécutez-le pendant un moment, l'effet est le suivant :

流的暂停和流动

我们从 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！

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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!