Node.js의 스트림 모듈(stream module)

1. 스트림이란

스트림(Stream)은 데이터를 청크(chunk) 단위로 처리하는 방식입니다. 대용량 파일이나 네트워크 데이터를 전체를 메모리에 로드하지 않고 조각조각 처리할 수 있어 메모리 효율이 높습니다. Node.js의 많은 내장 모듈(http, fs, zlib 등)이 스트림 인터페이스를 사용합니다.

2. 스트림의 종류

Node.js는 네 가지 종류의 스트림을 제공합니다.

• Readable: 데이터를 읽을 수 있는 스트림 (fs.createReadStream, http 요청)
• Writable: 데이터를 쓸 수 있는 스트림 (fs.createWriteStream, http 응답)
• Duplex: 읽기/쓰기 모두 가능한 스트림 (TCP 소켓)
• Transform: 데이터를 변환하면서 통과시키는 스트림 (zlib 압축)

3. Readable 스트림

3.1 파일 읽기 스트림

const fs = require('fs');

const readStream = fs.createReadStream('largefile.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024  // 64KB 청크
});

readStream.on('data', (chunk) => {
  console.log('청크 크기:', chunk.length);
  console.log('청크 내용:', chunk.substring(0, 100));
});

readStream.on('end', () => {
  console.log('파일 읽기 완료');
});

readStream.on('error', (err) => {
  console.error('에러:', err.message);
});

3.2 스트림 일시 정지/재개

const readStream = fs.createReadStream('file.txt');

readStream.on('data', (chunk) => {
  console.log('데이터 수신');

  // 일시 정지
  readStream.pause();

  setTimeout(() => {
    // 재개
    readStream.resume();
  }, 1000);
});

4. Writable 스트림

4.1 파일 쓰기 스트림

const fs = require('fs');

const writeStream = fs.createWriteStream('output.txt');

writeStream.write('첫 번째 줄\n');
writeStream.write('두 번째 줄\n');
writeStream.write('세 번째 줄\n');
writeStream.end('마지막 줄');  // end()로 스트림 종료

writeStream.on('finish', () => {
  console.log('파일 쓰기 완료');
});

writeStream.on('error', (err) => {
  console.error('에러:', err.message);
});

4.2 drain 이벤트

내부 버퍼가 가득 찼을 때 drain 이벤트를 활용합니다.

const fs = require('fs');

const writeStream = fs.createWriteStream('output.txt');

function writeData(data) {
  const canContinue = writeStream.write(data);

  if (!canContinue) {
    console.log('버퍼 가득 참, 대기 중...');
    writeStream.once('drain', () => {
      console.log('버퍼 비워짐, 계속 쓰기 가능');
    });
  }
}

for (let i = 0; i < 1000000; i++) {
  writeData(`데이터 라인 ${i}\n`);
}

반응형

5. pipe() 메서드

스트림을 연결하는 가장 간단한 방법입니다.

5.1 파일 복사

const fs = require('fs');

const readStream = fs.createReadStream('source.txt');
const writeStream = fs.createWriteStream('destination.txt');

readStream.pipe(writeStream);

writeStream.on('finish', () => {
  console.log('파일 복사 완료');
});

5.2 여러 스트림 연결

const fs = require('fs');
const zlib = require('zlib');

// 파일 읽기 → 압축 → 파일 쓰기
fs.createReadStream('file.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('file.txt.gz'))
  .on('finish', () => {
    console.log('압축 완료');
  });

// 압축 해제
fs.createReadStream('file.txt.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('file_unzipped.txt'));

5.3 HTTP 응답에 파일 스트리밍

const http = require('http');
const fs = require('fs');

const server = http.createServer((req, res) => {
  if (req.url === '/video') {
    const videoPath = 'video.mp4';
    const stat = fs.statSync(videoPath);

    res.writeHead(200, {
      'Content-Type': 'video/mp4',
      'Content-Length': stat.size
    });

    fs.createReadStream(videoPath).pipe(res);
  }
});

server.listen(3000);

6. Transform 스트림

데이터를 변환하면서 통과시키는 스트림입니다.

const { Transform } = require('stream');

// 대문자 변환 스트림
const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    const upperCased = chunk.toString().toUpperCase();
    callback(null, upperCased);
  }
});

// 사용
process.stdin
  .pipe(upperCaseTransform)
  .pipe(process.stdout);

6.1 커스텀 Transform 스트림

const { Transform } = require('stream');

class JSONParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();

    const lines = this.buffer.split('\n');
    this.buffer = lines.pop();  // 마지막 불완전한 줄 보관

    for (const line of lines) {
      if (line.trim()) {
        try {
          const obj = JSON.parse(line);
          this.push(obj);
        } catch (err) {
          callback(err);
          return;
        }
      }
    }

    callback();
  }
}

7. pipeline() 함수

Node.js 10부터 제공되는 안전한 스트림 연결 방식입니다. 에러 처리와 정리를 자동으로 해줍니다.

const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');

pipeline(
  fs.createReadStream('input.txt'),
  zlib.createGzip(),
  fs.createWriteStream('input.txt.gz'),
  (err) => {
    if (err) {
      console.error('파이프라인 에러:', err);
    } else {
      console.log('파이프라인 완료');
    }
  }
);

// Promise 버전 (Node.js 15+)
const { pipeline } = require('stream/promises');

async function compress() {
  await pipeline(
    fs.createReadStream('input.txt'),
    zlib.createGzip(),
    fs.createWriteStream('input.txt.gz')
  );
  console.log('압축 완료');
}

8. 스트림과 일반 방식 비교

const fs = require('fs');

// 일반 방식 - 전체 파일을 메모리에 로드
fs.readFile('largefile.txt', (err, data) => {
  // data에 전체 파일 내용이 로드됨 (메모리 사용량 높음)
  fs.writeFile('copy.txt', data, () => {});
});

// 스트림 방식 - 청크 단위로 처리
fs.createReadStream('largefile.txt')
  .pipe(fs.createWriteStream('copy.txt'));
// 청크 단위로 처리되어 메모리 효율적

결론

스트림은 대용량 데이터를 메모리 효율적으로 처리하는 Node.js의 핵심 기능입니다. Readable, Writable, Duplex, Transform 네 가지 종류가 있으며, pipe()나 pipeline()으로 스트림을 연결할 수 있습니다. 파일 처리, HTTP 응답, 데이터 압축 등에서 스트림을 활용하면 메모리 사용량을 크게 줄일 수 있습니다.