[Node.js][노드] stream(스트림) 이란??

Node.js에서 stream이란 "스트리밍 데이터로 작업하기 위한 추상적인 인터페이스"라고 공식 문서에 나와있다.

뭔 소리일까..?

일단 stream이란 개념을 짚고 넘어가야 될 듯 싶다.

stream이란 일종의 추상적인 개념인데 입출력 기기나 프로세스, 파일 등 어디로 가는 지, 어디로 나왔는 지 상관없이 통일된 방식으로 데이터를 다루기 위한 가상의 개념이다.

그러니까 stream을 정의하기란 너무나 모호하다.

Node.js에서 많은 Object들이 stream Object 이다. 예를 들어서 http 서버의 request나 process.stdout도 stream 인스턴스이다.

stream들은 읽을 수 있거나(readable), 쓸 수 있거나(writable) 혹은 둘 다(both)가 될 수 있다.

모든 stream들은 EventEmitter의 인스턴스이다. 따라서 이벤트 핸들러를 작성할 수 있다.

Stream 타입 종류

• Readable: 읽을 수 있는 스트림(ex. fs.createReadStream())

• Writable: 쓸 수 있는 스트림(ex. fs.createWriteStream())

• Duplex: 읽고 쓸 수 있는 스트림(ex. net.Socket)

• Transform: 데이터를 쓰고 읽을 때 데이터를 수정하거나 변형할 수 있는 Duplex 스트림(ex.zlib.createDeflate()) // 잘 모르겠지만 압축파일과 관련이 있는 듯

CreateWriteStream 예제 코드

const fs = require('fs');
const file = fs.createWriteStream('./test.txt');  // stream 객체로 생성

for(let i = 0; i <= 1e6; i++) {   // 1 * 10^6번동안 계속 write
    file.write('Hello World!!\n');  // 문자열을 스트림에 씀
}
file.end();     // stream에게 더 이상 데이터를 전달하지 않겠다는 메서드.

재밌는 건 다음과 같이 코딩했을 때이다.

const fs = require('fs');
const file = fs.createWriteStream('./test.txt');

for(let i = 0; i <= 5e6; i++) {
    file.write('Hello World.\n');
}
console.log("write end");
setTimeout(() => {
    console.log("end");
    file.end();
}, 5000);

작업관리자:

stream을 write 했을 때 메모리가 1GB 가까이 급격히 증가했던 점과 콘솔 창에 "end"가 적히는 시점부터 'test.txt' 파일 크기가 켜졌던 점, 마지막으로 end가 찍히고도 프로그램이 종료되지 않은 점이다.

차근차근 추리를 시작해보자.

• file.write를 했을 때는 스트림에 데이터가 누적되어서 메모리가 급격히 증가했던 것 같다.

• end가 적히는 시점, 즉 file.end()가 실행된 뒤부터 파일의 크기가 커졌던 점을 생각해보면 그때부터 file I/O 스레드가 돌면서 파일에 stream 데이터를 write 하기 시작했다는 점을 알 수 있다.

• really end가 찍히고도 끝나지 않았던 것은 file I/O가 돌고있는 스레드가 아직 완료되지 않았기 때문에 프로세스가 완전히 종료되지 않은 것으로 보인다.

이 부분에 대해선 공부해야될 부분이 많은 것 같다.

메모리가 증가한 것에 대해 이것저것 찾아보다가 다음과 같은 설명을 보았다.

file.write를 할 때 highWaterMark(내부 버퍼에 저장할 최대 바이트)에 지정한 값보다 큰 버퍼를 쓰려고 할 땐, write()가 false를 리턴한다.

스트림이 빠져나가는 동안 write()를 호출하면 청크가 버퍼링되고 false가 반환되고, 현재 버퍼링된 모든 청크가 빠져나가면 drain 이벤트가 발생한다.

만약 스트림이 다 빠져나가지 않았는데 write()가 계속 발생되면 메모리가 다시 반환되지 않을 수 있다.

그래서 메모리 누수가 발생했던 것이다.

메모리 누수를 막기위해서는 다음과 같이 write()가 false일 때 스트림에 모든 데이터가 빠져나간 뒤(drain 이벤트 발동) test 함수가 다시 동작할 수 있도록 만들어야 한다.(적합한 코드는 아니다.)

const fs = require('fs');
let file = fs.createWriteStream('./test.txt');
let ok = true;
let i = 0;
function test() {
    for(; i <= 2e6 && ok; i++) {
        ok = file.write('Hello World.\n');
        console.log(ok);
    }
    if(ok == false) {
        ok = true;
        file.once('drain', test);     // drain 이벤트가 발생했을 때 일회 리스너 함수 test를 등록
        console.log("call drain");
    }
    if(i == 2e6) {
        console.log(i,": end");
        file.end();
    }
}
test();

CreateReadStream 예제 코드

const fs = require('fs');
const file = fs.createReadStream('./test.txt');

let chunk;

file.on('readable', () => {   // stream에 데이터가 남아있는 경우 자동으로 발생하는 이벤트
    // 가끔 다 읽지 않았을 때 chunk가 null일 경우가 있다.
    while (null !== (chunk = file.read(13))) {        // read의 buffer 크기: 13
        console.log(chunk.toString(), "\nline");
    }
});
file.on('end', () => {    // stream에 데이터가 더 이상 남아있지 않는 경우 발생하는 이벤트
    console.log('end');
});

readable 이벤트는 Readable 스트림에 읽을 데이터가 남아있을 때 발생하는 이벤트이다.

이 이벤트 안에 read() 함수를 사용해서 데이터를 읽으면 된다.