Qu’est-ce qu’un flux inscriptible dans Nodejs ? Comment utiliser

Recommandations associées : "Tutoriel Nodejs"

Qu'est-ce qu'un flux inscriptible

Inscriptible A Le flux est une abstraction des données circulant vers un périphérique. Il est utilisé pour consommer les données circulant en amont. Les données peuvent être écrites sur le périphérique via un programme de flux inscriptible. Les fichiers de disque local ou les réponses réseau telles que TCP et HTTP sont couramment utilisés.

Regardez un exemple qui a été utilisé auparavant

process.stdin.pipe(process.stdout);

*process.stdout* est un flux inscriptible Le programme écrit les données transmises par le processus de flux lisible. stdin. périphérique de sortie standard d’entrée. Il est très simple de comprendre le flux inscriptible sur la base de la compréhension du flux lisible. Un flux est constitué de données directionnelles, où le flux lisible est la source de données, le flux inscriptible est la destination et le lien pipeline au milieu est bidirectionnel. flux.

Un flux inscriptible utilise

pour appeler la méthode **write() ** d'une instance de flux inscriptible afin d'écrire des données dans un flux inscriptible

const fs = require('fs');
const rs = fs.createReadStream('./w.js');
const ws = fs.createWriteStream('./copy.js');

rs.setEncoding('utf-8');
rs.on('data', chunk => {
  ws.write(chunk);
});

Comme mentionné précédemment, l'écoute de l'événement data du flux lisible fera passer le flux lisible en mode fluide. Nous appelons la méthode write() du flux inscriptible dans l'événement de rappel, afin que les données soient écrites dans le flux inscriptible. stream. Dans le périphérique abstrait, c'est-à-dire le fichier copy.js dans le répertoire actuel.

La méthode write() a trois paramètres

• chunk {String| Buffer}, qui représente les données à écrire
• encodage Lorsque les données écrites sont une chaîne, vous pouvez définir l'encodage
• rappel La fonction de rappel après l'écriture des données

Les flux inscriptibles personnalisés

sont similaires aux flux lisibles personnalisés. Les flux inscriptibles personnalisés simples ne nécessitent que deux étapes

1. Hériter du Inscriptible du module de flux La classe
implémente la méthode
2. _write()

Implémentons un flux inscriptible simple et convertissons les données transmises dans le flux inscriptible en Après la capitalisation, sortie vers le périphérique de sortie standard (un meilleur exemple pourrait être l'écriture sur un fichier de disque local, mais cela implique trop d'opérations fs, ce qui est gênant, alors soyez paresseux. L'écriture sur le périphérique de sortie standard est également un comportement d'écriture)

const Writable = require('stream').Writable

class OutputStream extends Writable {
    _write(chunk, enc, done) {
        // 转大写之后写入标准输出设备
        process.stdout.write(chunk.toString().toUpperCase());
        // 此处不严谨，应该是监听写完之后才调用 done
        process.nextTick(done);
    }
}

module.exports = OutputStream;

est la même que la méthode write() exposée par le flux inscriptible final. La méthode _write() a trois paramètres et a une fonction similaire

• chunk données écrites, large Une partie du temps, il s'agit d'un tampon, à moins que decodeStrings soit défini sur false
• encoding Si les données sont une chaîne, vous pouvez définir l'encodage, le tampon ou le mode objet ignorera le
• rappel La fonction de rappel après l'écriture des données peut notifier le flux des données suivantes ; lorsqu'une erreur se produit, vous pouvez également définir un paramètre d'erreur

Bien sûr , il existe également une méthode _writev() Implémentation, cette méthode n'est appelée que par la file d'attente d'écriture bloquée et n'a pas besoin d'être implémentée.

Instancier un flux inscriptible

Après avoir la classe de flux inscriptible, nous pouvons l'instancier et l'utiliser. Il existe plusieurs options lors de l'instanciation d'un flux inscriptible Select, le comprendre peut nous aider à comprendre les connaissances que nous utiliserons plus tard

• objectModeLa valeur par défaut est false Après l'avoir définie sur true, la méthode writable.write() le fait. non seulement écrire une chaîne En plus des tampons, des objets JavaScript arbitraires peuvent être écrits. Une option très utile, qui sera présentée en détail plus tard lorsque nous présenterons le flux de transformation
• highWaterMarkLa quantité maximale de données écrites à chaque fois, la valeur par défaut pour Le tampon est de 16 Ko, objectMode La valeur par défaut est 16
• decodeStringsS'il faut convertir les données entrantes en Buffer, la valeur par défaut est vraie

De cette façon, il sera plus clair de connaître la signification des paramètres passés dans la méthode _write(), et cela sera utile pour comprendre le mécanisme de contre-pression introduit plus tard.

Événements

Comme les flux lisibles, les flux inscriptibles ont également plusieurs événements couramment utilisés. Avec la base des flux lisibles, il est relativement simple à comprendre

pipe
• Lorsque le flux lisible appelle la méthode pipe() pour transmettre des données au flux inscriptible, l'événement pipe du flux inscriptible sera déclenché
unpipe
• Lorsque le flux lisible appelle la méthode unpipe() pour supprimer le transfert de données, l'événement unpipe du flux inscriptible sera déclenché

Ces deux événements sont utilisés pour notifier les données du flux inscriptibles viendront et seront coupées, sont rarement utilisées dans des situations normales. La méthode

writeable.write()

a une valeur de retour bool comme mentionné précédemment, highWaterMark, lorsque les données à écrire sont supérieures au highWaterMark du. stream inscriptible, les données ne seront pas écrites en même temps et certaines données seront conservées. À ce moment, writeable.write() renverra false, et si elle peut être traitée, elle renverra true<.>

drain

Déclenché lorsqu'il y avait des données bloquées auparavant, c'est-à-dire que writeable.write() a renvoyé false Après une période de digestion, l'arriéré de données a été traité et de nouvelles données. peut être continué à être écrit (le sens originel de drain) C'est-à-dire drainage et épuisement, assez vivant)

除了 write() 方法可写流还有一个常用的方法 end()，参数和 write() 方法相同，但也可以不传入参数，表示没有其它数据需要写入，可写流可以关闭了。

finish 当调用 writable.end() 方法，并且所有数据都被写入底层后会触发 finish 事件

同样出现错误后会触发 error 事件

back pressure

了解了这些事件，结合上之前提到的可读流的一些知识，我们就能探讨一些有意思的话题了。在最开始我们提到过用流相对于直接操作文件的好处之一是不会把内存压爆，那么流是怎么做到的呢？

最开始我们可能会想到因为流不是一次性把所有数据载入内存处理，而是一边读一边写。但我们知道一般读取的速度会远远快于写入的速度，那么 pipe()  方法是怎么做到供需平衡的呢？

回忆一些基础知识，我们自己来实现一下 pipe() 方法的核心原理

1. 可读流有流动和暂停两种模式，可以通过 **pause() 和 resume() **方法切换
2. 可写流的 **write() **方法会返回是否能处理当前的数据，每次可以处理多少是 hignWatermark 决定的
3. 当可写流处理完了积压数据会触发 drain 事件

我们可以利用这三点来做到数据读取和写入的同步，还是使用之前的例子，但为了使消费速度降下来，我们各一秒再通知完成

class OutputStream extends Writable {
    _write(chunk, enc, done) {
        // 转大写之后写入标准输出设备
        process.stdout.write(chunk.toString().toUpperCase());
        // 故意延缓通知继续传递数据的时间，造成写入速度慢的现象
        setTimeout(done, 1000);
    }
}

我们使用一下自定义的两个类

const RandomNumberStream = require('./RandomNumberStream');
const OutputStream = require('./OutputStream');

const rns = new RandomNumberStream(100);
const os = new OutputStream({
    highWaterMark: 8 // 把水位降低，默认16k还是挺大的
});

rns.on('data', chunk => {
    // 当待处理队列大于 highWaterMark 时返回 false
    if (os.write(chunk) === false) { 
        console.log('pause');
        rns.pause(); // 暂停数据读取
    }
});

// 当待处理队列小于 highWaterMark 时触发 drain 事件
os.on('drain', () => {
    console.log('drain')
    rns.resume(); // 恢复数据读取
});

结合前面的三点和注释很容易看懂上面代码，这就是 pipe() 方法起作用的核心原理。数据的来源的去向我们有了大概了解，后面可以开始介绍数据的加工

• duplex
• transform

更多编程相关知识，请访问：编程学习课程！！

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!