Angular - RxJS 란?

개념

RxJS는 Observables을 사용한 반응형 프로그래밍을 위한 라이브러리이다. 

이는 이벤트 처리와 데이터 스트림 관리를 위해 특히 비동기 프로그래밍에 널리 사용되는 JavaScript 라이브러리이다

 

비교

일반적인 자바스크립트 코드와의 비교

 

1) 자바스크립트

let count = 0;
let rate = 1000;
let lastClick = Date.now() - rate;
document.addEventListener('click', () => {
  if (Date.now() - lastClick >= rate) {
    console.log(`Clicked ${++count} times`);
    lastClick = Date.now();
  }
});

 

2) RxJS 사용 - 다양한 제어 연산자들은 이벤트 제어에 도움이됌 (제어 연산자 : filter , delay , debounceTime , 등)

import { fromEvent, throttleTime, scan } from 'rxjs';

fromEvent(document, 'click')
  .pipe(
    throttleTime(1000),
    scan((count) => count + 1, 0)
  )
  .subscribe((count) => console.log(`Clicked ${count} times`));

 이외에도 다양한 연산자는 아래를 참조

https://rxjs.dev/guide/operators

RxJS

 

특징

1. Observable

- 값 또는 이벤트의 호출 가능한 아이디어를 나타냄

- 일반적으로 subscribe() 메서드를 호출하여 관찰을 시작하고, next(), error(), complete() 메서드를 사용하여 데이터를 전달하고, 에러 및 완료를 처리한다. 이러한 메서드를 통해 데이터 스트림에서 발생하는 이벤트를 처리할 수 있다.

import { Observable } from 'rxjs';

const observable = new Observable((subscriber) => {
  subscriber.next(1);
  subscriber.next(2);
  subscriber.next(3);
  setTimeout(() => {
    subscriber.next(4);
    subscriber.complete();
  }, 1000);
});

console.log('just before subscribe');
observable.subscribe({
  next(x) {
    console.log('got value ' + x);
  },
  error(err) {
    console.error('something wrong occurred: ' + err);
  },
  complete() {
    console.log('done');
  },
});
console.log('just after subscribe');

2. Observer

- Observable이 전달한 값을 수신하는 방법을 알고 있는 콜백 모음

- Observable이 전달할 수 있는 각 알림 유형에 대해 하나씩 세 개의 콜백이 있는 객체

- Observable은 데이터 스트림을 발생시키는 제공자(provider)이고, Observer는 데이터 스트림을 사용하는 소비자(consumer)이다. Observable은 데이터를 발행(publish)하고, Observer는 데이터를 구독(subscribe)한다.

 

3. Subscription

- Observable의 실행을 나타냄, 기본적으로 unsubscribe()리소스를 해제하거나 Observable 실행을 취소하는 기능만 있음

import { interval } from 'rxjs';

const observable1 = interval(400);
const observable2 = interval(300);

const subscription = observable1.subscribe(x => console.log('first: ' + x));
const childSubscription = observable2.subscribe(x => console.log('second: ' + x));

subscription.add(childSubscription);

setTimeout(() => {
  // Unsubscribes BOTH subscription and childSubscription
  subscription.unsubscribe();
}, 1000);

4. Operators

 - map , filter, concat, reduce등과 같은 연산으로 처리하는 기능적 프로그래밍 스타일을 가능하게 하는 함수

 

5. Subject:

- EventEmitter와 동일하며 값이나 이벤트를 multiple Observers에게 멀티캐스팅하는 방법

import { Subject, from } from 'rxjs';

const subject = new Subject<number>();

subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
});
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`),
});

const observable = from([1, 2, 3]);

observable.subscribe(subject); // You can subscribe providing a Subject

 

6. Schedulers

- setTimeout 동시성을 제어하는 ​​중앙 집중식 디스패처로, 예를 들어 또는 기타 에서 계산이 발생할 때 조정할 수 있음

- 구독이 시작되는 시기와 알림이 전달되는 시기를 제어

- 현재 실행되는 실행 컨텍스트에서 작업을 예약하거나, 새로운 실행 컨텍스트에서 작업을 예약할 수 있다. 이를 통해 RxJS는 스레드 블로킹 없이 비동기 작업을 수행할 수 있다.

import { Observable, observeOn, asyncScheduler } from 'rxjs';

const observable = new Observable((proxyObserver) => {
  proxyObserver.next(1);
  proxyObserver.next(2);
  proxyObserver.next(3);
  proxyObserver.complete();
}).pipe(
  observeOn(asyncScheduler)
);

const finalObserver = {
  next(x) {
    console.log('got value ' + x);
  },
  error(err) {
    console.error('something wrong occurred: ' + err);
  },
  complete() {
    console.log('done');
  },
};

console.log('just before subscribe');
observable.subscribe(finalObserver);
console.log('just after subscribe');