NodeJS의 읽기 가능한 스트림 이해

관련 권장 사항: "node js tutorial"

읽기 가능한 스트림이란 무엇입니까

읽기 가능한 스트림은 프로그램 소비를 위해 데이터를 생성하는 스트림입니다. 일반적인 데이터 생성 방법에는 디스크 파일 읽기, 네트워크 요청 콘텐츠 읽기 등이 포함됩니다. 스트림 사용에 대한 이전 예를 살펴보세요.

const rs = fs.createReadStream(filePath);

rs는 읽기 가능한 스트림이고 데이터 생성 방법은 디스크 파일을 읽는 것입니다. , 우리의 공통 콘솔 process.stdin은 읽기 가능한 스트림이기도 합니다.

process.stdin.pipe(process.stdout);

간단한 문장으로 콘솔의 입력을 인쇄할 수 있습니다. process.stdin이 데이터를 생성하는 방식은 콘솔에서 사용자의 입력을 읽는 것입니다.

읽기 가능한 스트림에 대한 정의를 다시 살펴보겠습니다. 읽기 가능한 스트림은 프로그램 사용을 위한 데이터를 생성하는 스트림입니다.

사용자 정의 읽을 수 있는 스트림

시스템이 제공하는 것 외에도fs.CreateReadStream, 우리는 gulp 또는 Vinyl-fs에서 제공하는 src 방법도 자주 사용합니다.

gulp.src(['*.js', 'dist/**/*.scss'])

특정 방식으로 데이터를 생성하려면 프로그램 소비, 그래서 시작하는 방법?

간단한 두 단계

1. Readable 클래스를 상속받으세요
2. 스트림 모듈의 _read 메서드를 재정의하고 this.push를 호출하여 생성된 데이터를 읽을 수 있는 대기열에 넣으세요

Readable 클래스 읽기 가능한 스트림이 수행해야 할 대부분의 작업이 완료되었습니다. 이를 상속한 다음 _read 메서드에 데이터를 생성하는 메서드를 작성하여 사용자 정의 읽기 가능한 스트림을 구현하기만 하면 됩니다.

100밀리초마다 난수를 생성하는 스트림을 구현하려는 경우(별로 유용하지 않음)

const Readable = require('stream').Readable;

class RandomNumberStream extends Readable {
    constructor(max) {
        super()
    }

    _read() {
        const ctx = this;
        setTimeout(() => {
            const randomNumber = parseInt(Math.random() * 10000);

            // 只能 push 字符串或 Buffer，为了方便显示打一个回车
            ctx.push(`${randomNumber}\n`);
        }, 100);
    }
}

module.exports = RandomNumberStream;

코드의 클래스 상속 부분은 매우 간단합니다. 주로 _read 메소드의 구현을 살펴보면 몇 가지가 있습니다. 주목할 가치가 있습니다

1. 기본적으로 클래스에 _read 메소드가 구현되어 있지만 우리가 하는 일은 재정의하고 다시 쓰는 것입니다. _read 메소드에는 데이터 양을 지정하는 데 사용되는 매개변수 크기가 있습니다. 읽어서 read 메소드로 반환해야 하지만 이는 단지 참조 데이터일 뿐입니다. 많은 구현에서는 이 매개변수를 무시하고 여기서도 무시합니다. 나중에 자세히 설명하겠습니다
2. this.push를 통해 데이터를 버퍼에 푸시합니다. 버퍼 개념은 나중에 언급할 예정이며 일시적으로 수도관에 압착되어 소비할 준비가 된 것으로 이해됩니다. 푸시의 내용은 숫자가 아닌 문자열 또는 버퍼만 될 수 있습니다. 푸시 메서드에는 두 번째 매개변수 인코딩이 있습니다.

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

const rns = new RandomNumberStream();

rns.pipe(process.stdout);

3. 이것은 작동합니다. 콘솔에 숫자가 계속 표시되는 것을 보면 난수를 생성하는 읽기 가능한 스트림이 있습니다. 아직 해결해야 할 몇 가지 작은 문제
4. 정지하는 방법

100밀리초마다 숫자를 버퍼에 푸시하므로 로컬 파일을 읽는 것과 마찬가지로 읽기가 끝나면 데이터를 중지하는 방법을 나타냅니다. 읽혔어?

버퍼에 null을 푸시하면 됩니다. 소비자가 필요한 난수 수를 정의할 수 있도록 코드를 수정해 보겠습니다.

const Readable = require('stream').Readable;

class RandomNumberStream extends Readable {
    constructor(max) {
        super()
        this.max = max;
    }

    _read() {
        const ctx = this;

        setTimeout(() => {
            if (ctx.max) {
                const randomNumber = parseInt(Math.random() * 10000);

                // 只能 push 字符串或 Buffer，为了方便显示打一个回车
                ctx.push(`${randomNumber}\n`);
                ctx.max -= 1;
            } else {
                ctx.push(null);
            }
        }, 100);
    }
}

module.exports = RandomNumberStream;

최대 식별자를 사용하여 소비자가 필요한 문자 수를 지정할 수 있으며 이는 인스턴스화 중에 지정할 수 있습니다.

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

const rns = new RandomNumberStream(5);

rns.pipe(process.stdout);

보시다시피 콘솔 5자만 인쇄됩니다

setInterval 대신 setTimeout이 필요한 이유

주의 깊은 학생들은 100밀리초마다 난수를 생성할 때 setInterval을 호출하지 않고 setTimeout을 사용한다는 것을 알아차렸을 것입니다. 제작이 지연되어 반복되지 않았는데 결과는 맞나요?

이를 위해서는 스트림이 작동하는 두 가지 방식을 이해해야 합니다.

플로우 모드: 기본 시스템에서 데이터를 읽고 가능한 한 빨리 애플리케이션에 제공합니다.일시 중지 모드: 읽으려면 read() 메서드를 명시적으로 호출해야 합니다. 여러 데이터 블록

스트림은 기본적으로 일시 중지 모드에 있습니다. 이는 프로그램이 명시적으로 read() 메서드를 호출해야 함을 의미합니다. 그러나 이 예에서는 스트림이 파이프()를 전달하므로 호출하지 않고도 데이터를 가져올 수 있습니다. 메소드는 흐름 모드로 전환되어 데이터를 읽을 때까지 _read() 메소드가 자동으로 반복적으로 호출되므로 각 _read() 메소드에서 데이터를 한 번만 읽으면 됩니다.

흐름 모드와 일시 중지 모드 간 전환
2. 다음과 같은 방법으로 흐름을 기본 일시 중지 모드에서 흐름 모드로 전환할 수 있습니다.

데이터 이벤트 리스너를 추가하여 데이터 모니터링을 시작합니다

resume() 호출 데이터 스트림을 시작하는 메서드pipe() 메서드를 호출하여 데이터를 쓰기 가능한 다른 스트림으로 전송합니다

흐름 모드에서 일시 중지 모드로 전환하는 방법에는 두 가지가 있습니다.

1. 在流没有 pipe() 时，调用 pause() 方法可以将流暂停
2. pipe() 时，需要移除所有 data 事件的监听，再调用 unpipe() 方法

data 事件

使用了 pipe() 方法后数据就从可读流进入了可写流，但对我们好像是个黑盒，数据究竟是怎么流向的呢？我们看到切换流动模式和暂停模式的时候有两个重要的名词

1. 流动模式对应的 data 事件
2. 暂停模式对应的 read() 方法

这两个机制是我们能够驱动数据流动的原因，先来看一下流动模式 data 事件，一旦我们监听了可读流的 data 时、事件，流就进入了流动模式，我们可以改写一下上面调用流的代码

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

const rns = new RandomNumberStream(5);

rns.on('data', chunk => {
  console.log(chunk);
});

这样我们可以看到控制台打印出了类似下面的结果

<buffer>
<buffer>
<buffer>
<buffer>
<buffer></buffer></buffer></buffer></buffer></buffer>

当可读流生产出可供消费的数据后就会触发 data 事件，data 事件监听器绑定后，数据会被尽可能地传递。data 事件的监听器可以在第一个参数收到可读流传递过来的 Buffer 数据，这也就是我们打印的 chunk，如果想显示为数字，可以调用 Buffer 的 toString() 方法。

当数据处理完成后还会触发一个 end 事件，应为流的处理不是同步调用，所以如果我们希望完事后做一些事情就需要监听这个事件，我们在代码最后追加一句：

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

这样可以在数据接收完了显示 'done'

当然数据处理过程中出现了错误会触发 error 事件，我们同样可以监听，做异常处理：

rns.on('error', (err) => {
  console.log(err);
});

read(size)

流在暂停模式下需要程序显式调用 read() 方法才能得到数据。read() 方法会从内部缓冲区中拉取并返回若干数据，当没有更多可用数据时，会返回null。

使用 read() 方法读取数据时，如果传入了 size 参数，那么它会返回指定字节的数据；当指定的size字节不可用时，则返回null。如果没有指定size参数，那么会返回内部缓冲区中的所有数据。

现在有一个矛盾了，在流动模式下流生产出了数据，然后触发 data 事件通知给程序，这样很方便。在暂停模式下需要程序去读取，那么就有一种可能是读取的时候还没生产好，如果我们才用轮询的方式未免效率有些低。

NodeJS 为我们提供了一个 readable 的事件，事件在可读流准备好数据的时候触发，也就是先监听这个事件，收到通知又数据了我们再去读取就好了：

const rns = new RandomNumberStream(5);

rns.on('readable', () => {
  let chunk;
  while((chunk = rns.read()) !== null){
    console.log(chunk);
  }
});

这样我们同样可以读取到数据，值得注意的一点是并不是每次调用 read() 方法都可以返回数据，前面提到了如果可用的数据没有达到 size 那么返回 null，所以我们在程序中加了个判断。

数据会不会漏掉

开始使用流动模式的时候我经常会担心一个问题，上面代码中可读流在创建好的时候就生产数据了，那么会不会在我们绑定 readable 事件之前就生产了某些数据，触发了 readable 事件，我们还没有绑定，这样不是极端情况下会造成开头数据的丢失嘛

可事实并不会，按照 NodeJS event loop 我们创建流和调用事件监听在一个事件队列里面，儿生产数据由于涉及到异步操作，已经处于了下一个事件队列，我们监听事件再慢也会比数据生产块，数据不会丢失。

看到这里，大家其实对 data事件、readable事件触发时机， read() 方法每次读多少数据，什么时候返回 null 还有又一定的疑问，因为到现在为止我们接触到的仍然是一个黑盒，后面我们介绍了可写流后会在 back pressure 机制部分对这些内部细节结合源码详细讲解，且听下回分解吧。

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

위 내용은 NodeJS의 읽기 가능한 스트림 이해의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!