Qu'est-ce qu'un flux ? Comment comprendre les flux dans Nodejs

Qu'est-ce que le flux ? Comment comprendre le flux ? L'article suivant vous donnera une compréhension approfondie du Stream dans Node J'espère qu'il vous sera utile !

L'auteur a souvent utilisé la fonction pipe dans le développement récemment. Je sais seulement qu'il s'agit d'un pipeline pour les flux, mais je ne sais pas comment cela fonctionne. Donc, avec la mentalité de découvrir, je pars simplement de. le flux et apprenez-le avec désinvolture. Les connaissances et le code source que j'ai lu sont compilés dans un article à partager avec tout le monde.

Stream est un concept très basique dans Nodejs De nombreux modules de base sont implémentés sur la base de flux et jouent un rôle très important. Dans le même temps, le flux est également un concept très difficile à comprendre. Cela est principalement dû au manque de documentation pertinente. Pour les débutants en NodeJ, il faut souvent beaucoup de temps pour comprendre le flux avant de pouvoir réellement maîtriser ce concept. pour la plupart des NodeJ, c'est le cas. Pour les utilisateurs, il n'est utilisé que pour développer des applications Web. Une compréhension insuffisante des flux n'affecte pas leur utilisation. Cependant, comprendre les flux peut conduire à une meilleure compréhension des autres modules de NodeJs et, dans certains cas, l'utilisation de flux pour traiter les données fournira de meilleurs résultats. [Recommandations de didacticiel associées : Tutoriel vidéo Nodejs]

Comment comprendre les flux

• Pour les utilisateurs de flux, le flux peut être considéré comme un tableau, et nous devons uniquement nous concentrer sur l'obtention (la consommation) et l'écriture à partir de ça (Production) va bien.

• Pour les développeurs de flux (utilisant le module stream pour créer une nouvelle instance), ils se concentrent sur la façon d'implémenter certaines méthodes dans le flux. Ils se concentrent généralement sur deux points, qui est la ressource cible et comment exploiter la ressource cible. .Une fois déterminé, il est nécessaire d'exploiter les ressources cibles en fonction des différents états et événements du flux

Pool de cache

Tous les flux dans NodeJs ont des pools de tampons. Le but du pool de tampons est d'augmenter l'efficacité de. le flux lorsque les données sont produites et consommées. Lorsque tout cela prend du temps, nous pouvons produire des données à l'avance et les stocker dans le pool tampon avant la prochaine consommation. Cependant, le pool de mémoire tampon n'est pas toujours utilisé. Par exemple, lorsque le pool de cache est vide, les données ne seront pas placées dans le pool de cache après la production mais seront consommées directement. .

Si la vitesse de production des données est supérieure à la vitesse de consommation des données, les données excédentaires attendront quelque part. Si la vitesse de production des données est inférieure à la vitesse de consommation des données du processus, les données s'accumuleront jusqu'à une certaine quantité quelque part, puis seront consommées. (Les développeurs ne peuvent pas contrôler la vitesse de production et de consommation des données, ils peuvent seulement essayer de produire des données ou de consommer des données à quelle heure)

Cet endroit où les données attendent, accumulent des données, puis se produisent. C'est le pool tampon. Le pool de mémoire tampon est généralement situé dans la RAM (mémoire) de l'ordinateur.

Pour donner un exemple de tampon courant, lorsque nous regardons des vidéos en ligne, si votre vitesse Internet est rapide, le tampon sera toujours rempli immédiatement, puis envoyé au système pour lecture, puis la vidéo suivante sera immédiatement mise en mémoire tampon. Il n'y aura aucun décalage pendant la visualisation. Si la vitesse du réseau est très lente, vous verrez un chargement, indiquant que le tampon est en cours de remplissage. Une fois le remplissage terminé, les données sont envoyées au système et vous pouvez voir cette vidéo.

Le pool de cache du flux NodeJs est une liste chaînée Buffer. Chaque fois que vous souhaitez ajouter des données au pool de cache, un nœud Buffer sera recréé et inséré à la fin de la liste chaînée.

EventEmitter

Stream dans NodeJs est une interface abstraite qui implémente EventEmitter, je vais donc d'abord présenter brièvement EventEmitter.

EventEmitter est une classe qui implémente des fonctions de publication d'événements et d'abonnement. Plusieurs méthodes couramment utilisées (on, once, off, submit) sont considérées comme familières à tout le monde, je ne les présenterai donc pas une par une.

const { EventEmitter } = require('events')

const eventEmitter = new EventEmitter()

// 为 eventA 事件绑定处理函数
eventEmitter.on('eventA', () => {
    console.log('eventA active 1');
});

// 为 eventB 事件绑定处理函数
eventEmitter.on('eventB', () => {
    console.log('eventB active 1');
});

eventEmitter.once('eventA', () => {
    console.log('eventA active 2');
});

// 触发 eventA
eventEmitter.emit('eventA')
// eventA active 1
// eventA active 2

Il convient de noter que EventEmitter a deux événements appelés newListener et removeListener lorsque vous ajoutez un événement à un objet événement après avoir écouté. la fonction newListener (eventEmitter.emit('newListener')) sera déclenchée lorsqu'une fonction de traitement est supprimée, removeListener sera déclenchée de la même manière. EventEmitter 有两个叫做 newListener 和 removeListener 的事件，当你向一个事件对象中添加任何事件监听函数后，都会触发 newListener（eventEmitter.emit('newListener')），当一个处理函数被移除时同理会触发 removeListener。

还需要注意的是， once 绑定的处理函数只会执行一次，removeListener 将在其执行前被触发，这意味着 once

Il convient également de noter que la fonction de gestionnaire liée une fois ne sera exécutée qu'une seule fois, et removeListener sera déclenchée avant son exécution, ce qui signifie que l'écouteur lié par once La fonction est d'abord supprimée avant d'être déclenchée.

const { EventEmitter } = require('events')

const eventEmitter = new EventEmitter()

eventEmitter.on('newListener', (event, listener)=>{
    console.log('newListener', event, listener)
})

eventEmitter.on('removeListener', (event, listener) => {
    console.log('removeListener', event, listener)
})
//newListener removeListener[Function(anonymous)]


eventEmitter.on('eventA', () => {
    console.log('eventA active 1');
});
//newListener eventA [Function (anonymous)]

function listenerB() { console.log('eventB active 1'); }
eventEmitter.on('eventB', listenerB);
// newListener eventB [Function (anonymous)]

eventEmitter.once('eventA', () => {
    console.log('eventA active 2');
});
// newListener eventA [Function (anonymous)]

eventEmitter.emit('eventA')
// eventA active 1
// removeListener eventA [Function: bound onceWrapper] { listener: [Function (anonymous)] }
// eventA active 2

eventEmitter.off('eventB', listenerB)
// removeListener eventB[Function: listenerB]

Mais cela n'a pas d'importance pour ce que nous verrons plus tard.

Stream

Stream est une interface abstraite pour traiter les données en streaming dans Node.js. Stream n'est pas une interface réelle, mais un terme général désignant tous les flux. Les interfaces réelles sont ReadableStream, WritableStream et ReadWriteStream.

interface ReadableStream extends EventEmitter {
    readable: boolean;
    read(size?: number): string | Buffer;
    setEncoding(encoding: BufferEncoding): this;
    pause(): this;
    resume(): this;
    isPaused(): boolean;
    pipe<T extends WritableStream>(destination: T, options?: { end?: boolean | undefined; }): T;
    unpipe(destination?: WritableStream): this;
    unshift(chunk: string | Uint8Array, encoding?: BufferEncoding): void;
    wrap(oldStream: ReadableStream): this;
    [Symbol.asyncIterator](): AsyncIterableIterator<string | Buffer>;
}

interface WritableStream extends EventEmitter {
    writable: boolean;
    write(buffer: Uint8Array | string, cb?: (err?: Error | null) => void): boolean;
    write(str: string, encoding?: BufferEncoding, cb?: (err?: Error | null) => void): boolean;
    end(cb?: () => void): this;
    end(data: string | Uint8Array, cb?: () => void): this;
    end(str: string, encoding?: BufferEncoding, cb?: () => void): this;
}

interface ReadWriteStream extends ReadableStream, WritableStream { }

On peut voir que ReadableStream et WritableStream sont toutes deux des interfaces qui héritent de la classe EventEmitter (les interfaces dans ts peuvent hériter de classes, car ce ne sont que des types de fusion).

Les classes d'implémentation correspondant aux interfaces ci-dessus sont respectivement Readable, Writable et Duplex

Il existe 4 types de flux dans NodeJs : 🎜

• Flux lisible en lecture (implémente ReadableStream)
• Flux inscriptible en écriture (implémente WritableStream)
• Flux duplex lisible et inscriptible (implémente WritableStream après avoir hérité de Readable)
• Transformer le flux de conversion (hérite de Duplex)

Problème de contre-pression

Le la vitesse d'écriture des données sur le disque est bien inférieure à celle de la mémoire. Nous imaginons qu'il y a un « tuyau » entre la mémoire et le disque, et le « tuyau » est le « flux » dans lequel les données de la mémoire affluent. le tuyau très rapidement. Lorsque le tuyau est bloqué Lorsqu'il est plein, une contre-pression des données se produira dans la mémoire et les données seront accumulées dans la mémoire, occupant des ressources.

La solution pour NodeJs Stream est de définir une valeur flottante pour le pool de cache (c'est-à-dire la file d'attente d'écriture dans l'image) de chaque flux lorsque la quantité de données atteint ce flottant. valeur, accédez au cache Lorsque le pool push à nouveau les données, il renverra false, ce qui signifie que le contenu du pool de cache dans le flux actuel a atteint la valeur flottante et qu'aucune donnée supplémentaire ne devrait être écrit. À ce stade, nous devons arrêter immédiatement la production de données pour empêcher la mise en cache. Un pool surdimensionné crée une contre-pression. 缓存池（就是图中写入队列）设置一个浮标值，当其中数据量达到这个浮标值后，往缓存池再次 push 数据时就会返回 false，表示当前流中缓存池内容已经达到浮标值，不希望再有数据写入了，这时我们应该立即停止数据的生产，防止缓存池过大产生背压。

Readable

可读流（Readable）是流的一种类型，他有两种模式三种状态

两种读取模式：

• 流动模式：数据会从底层系统读取写入到缓冲区，当缓冲区被写满后自动通过 EventEmitter 尽快的将数据传递给所注册的事件处理程序中

• 暂停模式：在这种模式下将不会主动触发 EventEmitter 传输数据，必须显示的调用 Readable.read() 方法来从缓冲区中读取数据，read 会触发响应到 EventEmitter 事件。

三种状态：

• readableFlowing === null（初始状态）

• readableFlowing === false（暂停模式）

• readableFlowing === true（流动模式）

初始时流的 readable.readableFlowing 为 null

添加data事件后变为 true 。调用 pause()、unpipe()、或接收到背压或者添加 readable 事件，则 readableFlowing 会被设为 false ，在这个状态下，为 data 事件绑定监听器不会使 readableFlowing 切换到 true。

调用 resume() 可以让可读流的 readableFlowing

Readable

Readable stream (Readable) est un type de flux Il a deux modes et trois états

Deux modes de lecture :

Mode flux : les données seront lues et écrites du système sous-jacent dans le tampon. Lorsque le tampon est plein, les données seront automatiquement transmises au gestionnaire d'événements enregistré via EventEmitter dès que possible. Trois états :
readableFlowing === null (état initial) readableFlowing === false (mode pause) readableFlowing === true (mode flux) Le readable.readableFlowing du flux initial est null et devient vrai après l'ajout de l'événement de données. Lorsque pause(), unpipe() est appelé, ou qu'une contre-pression est reçue ou qu'un événement readable est ajouté, readableFlowing</code > will Est défini sur false,<strong>Dans cet état, lier un écouteur à l'événement de données ne fera pas passer readableFlowing à true</strong>. </tr><tr>Appelez <code>resume() pour faire passer le readableFlowing du flux lisible à true Ce sera être déclenché à chaque fois que les données sont consommées. Le paramètre est les données consommées cette fois. Déclenché lorsque le flux est fermé. Déclenché lorsqu'une erreur se produit dans le flux. une longueur de taille. Renvoyer null signifie que les données actuelles sont inférieures à la taille. Sinon, les données consommées cette fois sont renvoyées. Lorsque la taille n'est pas transmise, cela signifie consommer toutes les données du pool de cache

	Mode Pause dans le programme : Dans ce mode, EventEmitter ne sera pas activement déclenché pour transmettre des données. La méthode Readable.read() doit être explicitement appelée pour lire les données du tampon de lecture des événements. sera déclenché en réponse à EventEmitter.
			Supprimer tous les événements lisibles est le seul moyen de rendre readableFlowing nul. Nom de l'événement

const fs = require('fs');

const readStreams = fs.createReadStream('./EventEmitter.js', {
    highWaterMark: 100// 缓存池浮标值
})

readStreams.on('readable', () => {
    console.log('缓冲区满了')
    readStreams.read()// 消费缓存池的所有数据，返回结果并且触发data事件
})


readStreams.on('data', (data) => {
    console.log('data')
})

https://github1s.com/nodejs/node/blob/v16.14.0/lib/internal/streams/readable.js#L527

当 size 为 0 会触发 readable 事件。

当缓存池中的数据长度达到浮标值 highWaterMark 后，就不会在主动请求生产数据，而是等待数据被消费后在生产数据

暂停状态的流如果不调用 read 来消费数据时，后续也不会在触发 data 和 readable，当调用 read 消费时会先判断本次消费后剩余的数据长度是否低于 浮标值，如果低于 浮标值 就会在消费前请求生产数据。这样在 read 后的逻辑执行完成后新的数据大概率也已经生产完成，然后再次触发 readable，这种提前生产下一次消费的数据存放在缓存池的机制也是缓存流为什么快的原因

流动状态下的流有两种情况

• 生产速度慢于消费速度时：这种情况下每一个生产数据后一般缓存池中都不会有剩余数据，直接将本次生产的数据传递给 data 事件即可（因为没有进入缓存池，所以也不用调用 read 来消费），然后立即开始生产新数据，待上一次数据消费完后新数据才生产好，再次触发 data ，一只到流结束。
• 生产速度快于消费速度时：此时每一次生产完数据后一般缓存池都还存在未消费的数据，这种情况一般会在消费数据时开始生产下一次消费的数据，待旧数据消费完后新数据已经生产完并且放入缓存池

他们的区别仅仅在于数据生产后缓存池是否还存在数据，如果存在数据则将生产的数据 push 到缓存池等待消费，如果不存在则直接将数据交给 data 而不加入缓存池。

值得注意的是当一个缓存池中存在数据的流从暂停模式进入的流动模式时，会先循环调用 read 来消费数据只到返回 null

暂停模式



暂停模式下，一个可读流读创建时，模式是暂停模式，创建后会自动调用 _read 方法，把数据从数据源 push 到缓冲池中，直到缓冲池中的数据达到了浮标值。每当数据到达浮标值时，可读流会触发一个 " readable " 事件，告诉消费者有数据已经准备好了，可以继续消费。

一般来说， 'readable' 事件表明流有新的动态：要么有新的数据，要么到达流的尽头。所以，数据源的数据被读完前，也会触发一次 'readable' 事件；

消费者 " readable " 事件的处理函数中，通过 stream.read(size) 主动消费缓冲池中的数据。

const { Readable } = require('stream')

let count = 1000
const myReadable = new Readable({
    highWaterMark: 300,
    // 参数的 read 方法会作为流的 _read 方法，用于获取源数据
    read(size) {
        // 假设我们的源数据上 1000 个1
        let chunk = null
        // 读取数据的过程一般是异步的，例如IO操作
        setTimeout(() => {
            if (count > 0) {
                let chunkLength = Math.min(count, size)
                chunk = '1'.repeat(chunkLength)
                count -= chunkLength
            }
            this.push(chunk)
        }, 500)
    }
})
// 每一次成功 push 数据到缓存池后都会触发 readable
myReadable.on('readable', () => {
    const chunk = myReadable.read()//消费当前缓存池中所有数据
    console.log(chunk.toString())
})

值得注意的是， 如果 read(size) 的 size 大于浮标值，会重新计算新的浮标值，新浮标值是size的下一个二次幂（size <= 2^n，n取最小值）

//  hwm 不会大于 1GB.
const MAX_HWM = 0x40000000;
function computeNewHighWaterMark(n) {
  if (n >= MAX_HWM) {
    // 1GB限制
    n = MAX_HWM;
  } else {
    //取下一个2最高幂，以防止过度增加hwm
    n--;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    n++;
  }
  return n;
}

流动模式



所有可读流开始的时候都是暂停模式，可以通过以下方法可以切换至流动模式：

• 添加 " data " 事件句柄；
• 调用 “ resume ”方法；
• 使用 " pipe " 方法把数据发送到可写流

流动模式下，缓冲池里面的数据会自动输出到消费端进行消费，同时，每次输出数据后，会自动回调 _read 方法，把数据源的数据放到缓冲池中，如果此时缓存池中不存在数据则会直接吧数据传递给 data 事件，不会经过缓存池；直到流动模式切换至其他暂停模式，或者数据源的数据被读取完了（ push(null) ）;

可读流可以通过以下方式切换回暂停模式：

• 如果没有管道目标，则调用 stream.pause() 。
• 如果有管道目标，则移除所有管道目标。调用 stream.unpipe() 可以移除多个管道目标。

const { Readable } = require('stream')

let count = 1000
const myReadable = new Readable({
    highWaterMark: 300,
    read(size) {
        let chunk = null
        setTimeout(() => {
            if (count > 0) {
                let chunkLength = Math.min(count, size)
                chunk = '1'.repeat(chunkLength)
                count -= chunkLength
            }
            this.push(chunk)
        }, 500)
    }
})

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

Writable

相对可读流来说，可写流要简单一些。



当生产者调用 write(chunk) 时，内部会根据一些状态（corked，writing等）选择是否缓存到缓冲队列中或者调用 _write，每次写完数据后，会尝试清空缓存队列中的数据。如果缓冲队列中的数据大小超出了浮标值（highWaterMark），消费者调用 write(chunk) 后会返回 false，这时候生产者应该停止继续写入。

那么什么时候可以继续写入呢？当缓冲中的数据都被成功 _write 之后，清空了缓冲队列后会触发 drain 事件，这时候生产者可以继续写入数据。

当生产者需要结束写入数据时，需要调用 stream.end 方法通知可写流结束。

const { Writable, Duplex } = require('stream')
let fileContent = ''
const myWritable = new Writable({
    highWaterMark: 10,
    write(chunk, encoding, callback) {// 会作为_write方法
        setTimeout(()=>{
            fileContent += chunk
            callback()// 写入结束后调用
        }, 500)
    }
})

myWritable.on('close', ()=>{
    console.log('close', fileContent)
})
myWritable.write('123123')// true
myWritable.write('123123')// false
myWritable.end()

注意，在缓存池中数据到达浮标值后，此时缓存池中可能存在多个节点，在清空缓存池的过程中（循环调用_read）,并不会向可读流一样尽量一次消费长度为浮标值的数据，而是每次消费一个缓冲区节点，即使这个缓冲区长度于浮标值不一致也是如此

const { Writable } = require('stream')


let fileContent = ''
const myWritable = new Writable({
    highWaterMark: 10,
    write(chunk, encoding, callback) {
        setTimeout(()=>{
            fileContent += chunk
            console.log('消费', chunk.toString())
            callback()// 写入结束后调用
        }, 100)
    }
})

myWritable.on('close', ()=>{
    console.log('close', fileContent)
})

let count = 0
function productionData(){
    let flag = true
    while (count <= 20 && flag){
        flag = myWritable.write(count.toString())
        count++
    }
    if(count > 20){
        myWritable.end()
    }
}
productionData()
myWritable.on(&#39;drain&#39;, productionData)

上述是一个浮标值为 10 的可写流，现在数据源是一个 0——20 到连续的数字字符串，productionData 用于写入数据。

• 首先第一次调用 myWritable.write("0") 时，因为缓存池不存在数据，所以 "0" 不进入缓存池，而是直接交给 _wirte，myWritable.write("0") 返回值为 true

• 当执行 myWritable.write("1") 时，因为 _wirte 的 callback 还未调用，表明上一次数据还未写入完，位置保证数据写入的有序性，只能创建一个缓冲区将 "1" 加入缓存池中。后面 2-9 都是如此

• 当执行 myWritable.write("10") 时，此时缓冲区长度为 9（1-9），还未到达浮标值， "10" 继续作为一个缓冲区加入缓存池中，此时缓存池长度变为 11，所以 myWritable.write("1") 返回 false，这意味着缓冲区的数据已经足够，我们需要等待 drain 事件通知时再生产数据。

• 100ms过后，_write("0", encoding, callback) 的 callback 被调用，表明 "0" 已经写入完成。然后会检查缓存池中是否存在数据，如果存在则会先调用 _read 消费缓存池的头节点("1")，然后继续重复这个过程直到缓存池为空后触发 drain 事件，再次执行 productionData

• 调用 myWritable.write("11")，触发第1步开始的过程，直到流结束。

Duplex

在理解了可读流与可写流后，双工流就好理解了，双工流事实上是继承了可读流然后实现了可写流（源码是这么写的，但是应该说是同时实现了可读流和可写流更加好）。



Duplex 流需要同时实现下面两个方法

• 实现 _read() 方法，为可读流生产数据

• 实现 _write() 方法，为可写流消费数据

上面两个方法如何实现在上面可写流可读流的部分已经介绍过了，这里需要注意的是，双工流是存在两个独立的缓存池分别提供给两个流，他们的数据源也不一样

以 NodeJs 的标准输入输出流为例：

• 当我们在控制台输入数据时会触发其 data 事件，这证明他有可读流的功能，每一次用户键入回车相当于调用可读的 push 方法推送生产的数据。
• 当我们调用其 write 方法时也可以向控制台输出内容，但是不会触发 data 事件，这说明他有可写流的功能，而且有独立的缓冲区，_write 方法的实现内容就是让控制台展示文字。

// 每当用户在控制台输入数据（_read），就会触发data事件，这是可读流的特性
process.stdin.on('data', data=>{
    process.stdin.write(data);
})

// 每隔一秒向标准输入流生产数据（这是可写流的特性，会直接输出到控制台上），不会触发data
setInterval(()=>{
    process.stdin.write('不是用户控制台输入的数据')
}, 1000)

Transform



可以将 Duplex 流视为具有可写流的可读流。两者都是独立的，每个都有独立的内部缓冲区。读写事件独立发生。

                             Duplex Stream
                          ------------------|
                    Read  <-----               External Source
            You           ------------------|  
                    Write ----->               External Sink
                          ------------------|

Transform 流是双工的，其中读写以因果关系进行。双工流的端点通过某种转换链接。读取要求发生写入。

                                 Transform Stream
                           --------------|--------------
            You     Write  ---->                   ---->  Read  You
                           --------------|--------------

对于创建 Transform 流，最重要的是要实现 _transform 方法而不是 _write 或者 _read。 _transform 中对可写流写入的数据做处理（消费）然后为可读流生产数据。

转换流还经常会实现一个 `_flush` 方法，他会在流结束前被调用，一般用于对流的末尾追加一些东西，例如压缩文件时的一些压缩信息就是在这里加上的

const { write } = require('fs')
const { Transform, PassThrough } = require('stream')

const reurce = '1312123213124341234213423428354816273513461891468186499126412'

const transform = new Transform({
    highWaterMark: 10,
    transform(chunk ,encoding, callback){// 转换数据，调用push将转换结果加入缓存池
        this.push(chunk.toString().replace('1', '@'))
        callback()
    },
    flush(callback){// end触发前执行
        this.push(&#39;<<<&#39;)
        callback()
    }
})


// write 不断写入数据
let count = 0
transform.write(&#39;>>>&#39;)
function productionData() {
    let flag = true
    while (count <= 20 && flag) {
        flag = transform.write(count.toString())
        count++
    }
    if (count > 20) {
        transform.end()
    }
}
productionData()
transform.on(&#39;drain&#39;, productionData)


let result = &#39;&#39;
transform.on(&#39;data&#39;, data=>{
    result += data.toString()
})
transform.on(&#39;end&#39;, ()=>{
    console.log(result)
    // >>>0@23456789@0@1@2@3@4@5@6@7@8@920<<<
})

Pipe

管道是将上一个程序的输出作为下一个程序的输入，这是管道在 Linux 中管道的作用。NodeJs 中的管道其实也类似，它管道用于连接两个流，上游的流的输出会作为下游的流的输入。



管道 sourec.pipe(dest, options) 要求 sourec 是可读的，dest是可写的。其返回值是 dest。

对于处于管道中间的流既是下一个流的上游也是上一个流的下游，所以其需要时一个可读可写的双工流，一般我们会使用转换流来作为管道中间的流。

https://github1s.com/nodejs/node/blob/v17.0.0/lib/internal/streams/legacy.js#L16-L33

Stream.prototype.pipe = function(dest, options) {
  const source = this;

  function ondata(chunk) {
    if (dest.writable && dest.write(chunk) === false && source.pause) {
      source.pause();
    }
  }

  source.on('data', ondata);

  function ondrain() {
    if (source.readable && source.resume) {
      source.resume();
    }
  }

  dest.on('drain', ondrain);
  // ...后面的代码省略
}

pipe 的实现非常清晰，当上游的流发出 data 事件时会调用下游流的 write 方法写入数据，然后立即调用 source.pause() 使得上游变为暂停状态，这主要是为了防止背压。

当下游的流将数据消费完成后会调用 source.resume() 使上游再次变为流动状态。

我们实现一个将 data 文件中所有 1 替换为 @ 然后输出到 result 文件到管道。

const { Transform } = require('stream')
const { createReadStream, createWriteStream } = require('fs')

// 一个位于管道中的转换流
function createTransformStream(){
    return new Transform({
        transform(chunk, encoding, callback){
            this.push(chunk.toString().replace(/1/g, '@'))
            callback()
        }
    })
}
createReadStream('./data')
.pipe(createTransformStream())
.pipe(createWriteStream('./result'))

在管道中只存在两个流时，其功能和转换流有点类似，都是将一个可读流与一个可写流串联起来，但是管道可以串联多个流。

原文地址：https://juejin.cn/post/7077511716564631566

作者：月夕

更多node相关知识，请访问：nodejs 教程！

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!

Étiquettes associées：
nodejs node.js node
source：juejin.cn
Article précédent：Un article pour parler du cycle de vie en Angular Article suivant：Comment lancer un projet Angular ? Parlons du processus en ligne avec nginx !
Déclaration de ce site Web
Le contenu de cet article est volontairement contribué par les internautes et les droits d'auteur appartiennent à l'auteur original. Ce site n'assume aucune responsabilité légale correspondante. Si vous trouvez un contenu suspecté de plagiat ou de contrefaçon, veuillez contacter admin@php.cn
Derniers articles par auteur
• Comprendre la sentinelle dans Redis en profondeur
  2023-04-26 17:59:18
• [Organisation et partage] 7 outils de gestion d'état React populaires
  2023-04-26 17:47:48
• Un article discute de la différence entre la clé dans Vue2 et la clé dans Vue3
  2023-04-26 17:41:42
• Un article pour parler du contrôle de la mémoire dans Node
  2023-04-26 17:37:05
• Partager des compétences pratiques sur Excel : 4 astuces pour supprimer les valeurs en double !
  2023-04-26 17:31:25
• Partager des compétences pratiques sur Word : la fonction de conversion simplifié vers traditionnel peut être utilisée de cette manière !
  2023-04-26 17:27:32
• Comment résoudre les problèmes cross-domaines ? Une brève analyse des solutions courantes
  2023-04-25 19:57:58
• Un article pour comprendre le modèle singleton en JavaScript
  2023-04-25 19:53:11
• En savoir plus sur les tampons dans Node
  2023-04-25 19:49:11
• Découvrez comment écrire des tests unitaires dans Vue3
  2023-04-25 19:41:54
Derniers numéros
Une erreur se produit lors de l'utilisation de NodeJS pour se connecter à MySQL dans Docker J'ai créé un serveur backend NodeJS connecté à MySQL. À l'aide de Docker, j'ai créé une im...
Depuis 2024-04-06 12:07:19
0
1
415
React/Node : erreur d'intégration rencontrée lors de l'abonnement à l'aide de Trial_period_days J'ai un produit défini dans Stripe que je souhaite configurer comme un essai gratuit de 7 ...
Depuis 2024-04-04 20:30:57
0
1
320
next ne fonctionne pas correctement dans le middleware utilisant Nodejs Je travaille avec Nodejs et j'utilise expressjs et maintenant je travaille sur la fonction...
Depuis 2024-04-04 17:33:56
0
1
340
Un simple programme Hello World crée plusieurs instances de nœuds Je suis très nouveau sur Javascript et NodeJS. J'exécute un simple programme helloworld co...
Depuis 2024-04-04 11:03:22
0
1
310
La requête SQL n'apparaît pas sur un itinéraire spécifique dans Node.js Je suis nouveau sur javascript et nodejs et j'ai une question. J'ai un code dans lequel je...
Depuis 2024-04-04 00:24:22
0
1
344
Rubriques connexes
Plus>
• qu'est-ce que nodejs
• Nodejs implémente le robot d'exploration
• débogage node.js
• Où puis-je me procurer le costume Douyin Little Fire Man ?
• Adresse du portefeuille Yiooke
• Collection complète de codes de couleurs CSS
• securefx ne peut pas se connecter
• Tutoriel d'utilisation de Kindeditor
Recommandations populaires
• que signifie js
• Comment convertir une chaîne en tableau en js ?
• Comment actualiser la page en utilisant javascript
• Comment supprimer un élément dans le tableau js
• Comment utiliser la fonction sqrt
Tutoriels populaires
Plus>
Tutoriels associés
Recommandations populaires
Derniers cours
• 
  L'ensemble complet de didacticiels vidéo Node.JS le plus classique

  22137
• 
  4 jours pour maîtriser le tutoriel nodejs

  104394
• 
  Node.js, voici comment l'apprendre

  23269
• 
  Tutoriel de base de Node.js

  10467
• 
  Cours pratique sur le projet de blog personnel Node.js

  61073
• 
  Le chemin d'un ingénieur full-stack vers Node.js

  282428
• 
  Le dernier didacticiel vidéo ThinkPHP 5.1 en première mondiale (60 jours pour devenir un expert PHP en ligne)

  1415245 temps d'étude
• 
  Tutoriel vidéo JAVA pour débutants

  2461310 temps d'étude
• 
  Tutoriel vidéo d'introduction base zéro à l'apprentissage de Python de Little Turtle

  502580 temps d'étude
• 
  Introduction rapide au développement web front-end

  215159 temps d'étude
• 
  Maîtrisez les didacticiels vidéo PS à partir de zéro

  873026 temps d'étude
• 
  [Web front-end] Démarrage rapide de Node.js

  6092 temps d'étude
• 
  Collection complète de cours full-stack de développement Web étranger

  4650 temps d'étude
• 
  Aller au langage pratique GraphQL

  4004 temps d'étude
• 
  Le maître du ventilateur de 550 W apprend JavaScript à partir de zéro, étape par étape

  596 temps d'étude
• 
  Le maître Python Mosh, un débutant sans aucune connaissance de base peut commencer en 6 heures

  20414 temps d'étude
Derniers téléchargements
Plus>
effets Web
Code source du site Web
Matériel du site Web
Modèle frontal
• [bouton de formulaire] Code de contact du formulaire de message d'entreprise jQuery
• [Effets spéciaux du joueur] Effets de lecture de boîte à musique HTML5 MP3
• [Navigation dans les menus] Effets spéciaux du menu de navigation d'animation de particules cool HTML5
• [bouton de formulaire] Code d'édition par glisser-déposer du formulaire visuel jQuery
• [Effets spéciaux du joueur] Code du lecteur de musique Kugou imitation VUE.JS
• [effets spéciaux HTML5] Jeu de boîte de poussée HTML5 classique
• [Effets spéciaux d'image] défilement jQuery pour ajouter ou réduire des effets d'image
• [Effets d'album photo] Effet de zoom de survol de la couverture de l'album personnel CSS3
• [Modèle frontal] Modèle de site Web d'entreprise de services de nettoyage et de réparation de décoration intérieure
• [Modèle frontal] Modèle de page de guide de CV personnel aux couleurs fraîches
• [Modèle frontal] Modèle Web de CV de travail créatif de concepteur
• [Modèle frontal] Modèle de site Web d'entreprise de construction d'ingénierie moderne
• [Modèle frontal] Modèle HTML5 réactif pour les établissements de services éducatifs
• [Modèle frontal] Modèle de site Web de centre commercial de boutique de livres électroniques en ligne
• [Modèle frontal] La technologie informatique résout le modèle de site Web d'entreprise Internet
• [Modèle frontal] Modèle de site Web de service de trading de devises de style violet
• [PNG素材] 可爱的夏天元素矢量素材(EPS+PNG)
• [PNG素材] 四个红的的 2023 毕业徽章矢量素材(AI+EPS+PNG)
• [banner图] 唱歌的小鸟和装满花朵的推车设计春天banner矢量素材(AI+EPS)
• [PNG素材] 金色的毕业帽矢量素材(EPS+PNG)
• [PNG素材] 黑白风格的山脉图标矢量素材(EPS+PNG)
• [PNG素材] 不同颜色披风和不同姿势的超级英雄剪影矢量素材(EPS+PNG)
• [banner图] 扁平风格的植树节banner矢量素材(AI+EPS)
• [PNG素材] 九个漫画风格的爆炸聊天气泡矢量素材(EPS+PNG)
• [Modèle frontal] Modèle de site Web d'entreprise de services de nettoyage et de réparation de décoration intérieure
• [Modèle frontal] Modèle de page de guide de CV personnel aux couleurs fraîches
• [Modèle frontal] Modèle Web de CV de travail créatif de concepteur
• [Modèle frontal] Modèle de site Web d'entreprise de construction d'ingénierie moderne
• [Modèle frontal] Modèle HTML5 réactif pour les établissements de services éducatifs
• [Modèle frontal] Modèle de site Web de centre commercial de boutique de livres électroniques en ligne
• [Modèle frontal] La technologie informatique résout le modèle de site Web d'entreprise Internet
• [Modèle frontal] Modèle de site Web de service de trading de devises de style violet

À propos de nous Clause de non-responsabilité Sitemap

Site Web PHP chinois：Formation PHP en ligne sur le bien-être public，Aidez les apprenants PHP à grandir rapidement！