[RxJava] #1

안녕하세요 

오늘은 비동기 프로그래밍 중 하나인 RxJava에 대해 알아보겠습니다

API와 통신할 때만 사용했기에, 더 많은 기능을 알고자 정리해보겠습니당

 

우선, 명령형 프로그래밍과 반응형 프로그래밍에 대해 간략히 알아볼게요 

RxJava는 반응형 프로그래밍을 따릅니다 

📌 명령형 프로그래밍
      - 선언된 명령어를 순차적으로 실행 

📌 반응형 프로그래밍
      - 데이터의 변화에 반응해 실행 (비동기 스트림으로 처리)

키워드에 포커스를 맞추고자 하다보니 문장이 좀 어색하네요 😅

---

RxJava

 

Rx는 ObserverPattern을 따릅니다 

아래의 포스팅을 확인해주세요 

✅ 2022.07.21 - [💻 프로그래밍] - [디자인 패턴] Observer Pattern(관찰자 패턴)

[디자인 패턴] Observer Pattern(관찰자 패턴)

 

한 문장으로 보자면 RxJava는 

비동기 *이벤트 기반 반응형 프로그래밍으로,  관찰가능한(*Observable) 스트림을 사용합니다

하지만 RxJava를 제대로 알려면 위의 문장에도 하나하나 해석이 필요해요 

 

📎 이벤트 기반 : 발생하는 이벤트(onNext / onError / onComplete)를 observable에 Stream으로 전달

      • onNext: 데이터 발행  시 실행되는 이벤트 

      • onError: 오류 발생 시 실행되는 이벤트 

      • onComplete: onNext가 마지막으로 발행될 때 실행되는 이벤트로, 어떤 에러도 발생하지 않습니다 

📎 Observable: 발생한 이벤트를 Observer(관찰자)에게 전달합니다

 

---

 

RxJava 사용을 위해선 4가지의 개념은 알아야합니다 

 

1️⃣ Observable : 데이터, 이벤트의 변화를 스트림형태로 생성

2️⃣ Observer  : 1️⃣의 데이터, 이벤트 변화에 반응

3️⃣ Subscriber : 발생 데이터와 결과를 연결 

4️⃣ Schedulers :  Observable, Observer에게 실행될 스레드를 알려줍니다 

 

💡 RxJava의 기본구성

• 생산자(Publisher) : 데이터를 생성하고 통지 ex) Flowable(*배압 기능 O), Observable(배압 기능 X)

• 소비자(Subscriber) : 데이터를 받아서 처리 ex) Subscriber, Observer

🔍 배압(Back Pressure)이란?
     데이터를 생성하는 속도와 데이터를 소비하는 속도의 차이가 큰 것을 말합니다  
     ex) 데이터를 생성하는 속도가 0.1초이고 처리하는 속도가 10초라고 한다면, 
           데이터가 쌓이기만 해서 메모리는 overflow가 되고 OOM이 발생합니다 

✏️ 배압 기능이 있다 -> 데이터 스트림에 쌓이는 데이터의 양을 제어 가능 

 

💡 Publisher의 종류

• Observable : 1개 이상의 데이터를 발행, 배압 기능 없음

• Flowable : 1개 이상의 데이터를 발행, 배압 기능 있음

• Single : 1개의 데이터를 발행

• Completable : 데이터 발행의 성공, 실패만을 나타냄 (onComplete / onError)

• Maybe : 1개의 데이터를 발행하지만, 발행하지 않을 수도 있음 (Single + Completable의 형태)

 

📍 Observable의 operator 

     observable을 새로 생성하거나, 변환 / stream을 변환할 때 사용됩니다

    operator는 용도가 다양한 만큼 종류가 엄청 많기 때문에 공식문서를 첨부합니다

 

💡 Observer

데이터, 이벤트 변화에 반응하는 인터페이스로, 데이터들을 수신합니다

 

💡 Subscriber

Observable에 Observer를 등록시킵니다 

 

💡 Scheduler

Thread를 분리해 뭘 처리할지를 정해줍니다 

subscribeOn : 생성자의 Thread를 지정

observeOn : 전달받은 데이터를 처리하는 Thread 지정(소비자의 Thread를 지정)

 

📎 Scheduler의 종류

•  Schdulers.io : Thread Pool에서 스레드를 가져오거나 쓰레드가 없으면 새로운 쓰레드를 생성

                               I/O 처리 작업, 네트워크 요청 처리, 데이터베이스쿼리 작업에 사용

•  Schdulers.computation : 간단한 계산, 연산 처리 

•  Schdulers.newThread : 매번 새로운 쓰레드를 생성

•  Schdulers.trampoline :  새로운 스레드를 생성하지 않고 사용하고 있는 현재 스레드에 무한한 크기의 대기 큐를 생성해 처리할 작업들을 FIFO로 처리

•  Schdulers.single :  단일 쓰레드를 생성, 여러번 구독해도 공통으로 사용

•  Schdulers.from(executer) :  Executor를 사용해서 생성한 쓰레드를 사용

 

 

---

 

장점

1. 다양한 비동기 처리 : 순차 / 다수의 비동기 등

2. 콜백제거 (unsubscriber호출로 인한 콜백해제, 모든 비동기 동작은 observable<T>타입)

3. 쓰레드관리 용이 (다양한 thread 관리 형식 지원)

4. 코드 깔끔 

 

단점

1. 높은 러닝커브 

2. 단순한 코드라면 오히려 Rx적용함이 더 복잡해지는 경우도 있다 

3. memory leak : subscriber과 함께 같이 사용되는 unsubscriber 

 

---

RxJava의 개념에 대해 알아봤습니다, 예제는 쭉 이어서 포스팅할게요 :) 

궁금하신 점이나 의견이 있으시면 댓글 부탁드립니다 감사합니다 😊