[Node.js] RxJS란

[Node.js] 스트림이란 (Stream) ①

 

[Node.js] 스트림이란 (Stream) ②

 

 이전 포스트에서 Stream을 다룬 이유는 RxJS를 접하게 되면서 Reactive Programming이라는 패러다임을 알게 되었고 막힘없으며 비동기적인 데이터 흐름을 공부하면서 작성하였습니다. 이번 글에서는 제가 찍먹 했던 RxJS에 대해 간략히 기록해두려고 합니다.
 포스팅 하단에는 RxJS를 학습하기 좋은 페이지의 링크를 첨부할테니 이론과 실습을 함께 공부해 보시면 좋을 것 같습니다.

 RxJS는 Reactive Extensions for JavaScript의 약어로, JavaScript에서 리액티브 프로그래밍을 구현하기 위한 라이브러리입니다. 리액티브 프로그래밍은 데이터의 흐름을 관찰하고, 이를 기반으로 비동기적인 데이터 처리를 하는 프로그래밍 패러다임으로, 이를 구현하기 위한 도구로 RxJS를 사용할 수 있습니다. RxJS는 옵저버 패턴, 이터레이터 패턴, 함수형 프로그래밍 등의 개념을 활용하며, 스트림(Stream)을 이용하여 비동기 데이터 처리를 구현할 수 있습니다. RxJS는 Angular 등의 프레임워크와 함께 많이 사용되며, 비동기 데이터 처리와 상태 관리 등에서 많은 도움을 줍니다.

 

 아래에는 RxJS를 이용해 Stream을 다루는 방법에 대해 간략히 적어보겠습니다.

 

스트림 생성

 먼저, 스트림을 생성하는 방법부터 알아봅시다. RxJS에서는 Observable이라는 클래스를 이용해 스트림을 생성할 수 있습니다. 아래 코드는 1초마다 1씩 증가하는 값을 방출하는 스트림을 생성하는 예시입니다.

import { Observable } from 'rxjs';

const stream$ = new Observable(observer => {
  let count = 0;

  const intervalId = setInterval(() => {
    observer.next(count++);
  }, 1000);

  return () => clearInterval(intervalId);
});

 여기서 stream$는 1초마다 값을 방출하는 스트림을 나타내는 객체입니다. new Observable() 생성자의 인자로 콜백 함수를 전달합니다. 이 콜백 함수는 스트림이 시작될 때 자동으로 호출되며, observer라는 인자를 받습니다. observer는 스트림에 값을 방출할 수 있는 next() 메서드와 스트림을 종료할 수 있는 complete() 메서드를 제공합니다.

 위 코드에서는 setInterval() 함수를 이용해 1초마다 count 값을 방출하고 있습니다. clearInterval() 함수를 이용해 스트림이 종료될 때 이 인터벌을 정리하고 있습니다.

 

스트림 구독

 이제, 생성한 스트림을 구독하는 방법을 살펴봅시다. 구독은 subscribe() 메소드를 이용해 이루어집니다. 아래 코드는 생성한 스트림을 구독하는 예시입니다.

stream$.subscribe({
  next: value => console.log(value),
  complete: () => console.log('stream completed'),
});

 위 코드에서는 next 콜백 함수를 이용해 스트림에서 방출되는 값을 받아와 출력하고 있습니다. complete 콜백 함수는 스트림이 종료될 때 호출됩니다.

 

연산자 사용

 RxJS는 다양한 연산자를 제공해 스트림의 값을 변환하거나 가공할 수 있습니다. 아래 코드는 map() 연산자를 이용해 스트림에서 방출되는 값을 두 배로 증가시키는 예시입니다.

stream$.pipe(
  map(value => value * 2),
).subscribe({
  next: value => console.log(value),
  complete: () => console.log('stream completed'),
});

 위 코드는 옵저버블 스트림을 생성하고, 해당 스트림을 가공하는 연산자 중 하나인 map() 함수를 적용하여, 스트림의 값에 2를 곱하는 과정을 수행하고 있습니다. map() 함수는 RxJS에서 제공하는 연산자 중 하나로, 입력 스트림을 받아서 변환된 출력 스트림을 생성하는 함수입니다.

 그 다음, subscribe() 함수를 통해 스트림에 옵저버를 등록하고, 등록된 옵저버는 next 콜백 함수를 이용하여, 스트림으로부터 발행되는 데이터를 처리하게 됩니다. 여기서는 console.log()를 이용하여, next 콜백 함수를 호출할 때마다 스트림으로부터 발행되는 값의 2배를 출력하게 됩니다. 또한, complete 콜백 함수를 이용하여, 스트림의 처리가 완료되면 해당 사실을 콘솔에 출력하게 됩니다.

---

 아래 링크는 "Learn RxJS"라는 RxJS 학습 사이트입니다. 웹에서 실습이 가능한 Stack Blitz와 연동이 잘 되어있어서 편리하게 실습을 해볼 수 있고 자주 사용하는 기능에 대해 마킹도 되었어서 학습의 장벽을 낮추는데 많은 도움이 될 것 같습니다.

Introduction - Learn RxJS

 

RxJS Docs - https://rxjs.dev/

npm - https://www.npmjs.com/package/rxjs

git repo - https://github.com/reactivex/rxjs