[RxJava] RxJava 빠르게 훑어보기

Reactive Programming

RxJava에 대해 이해하려면 Reactive Programming 에 대한 이해가 먼저 필요하다.

Reactive Programming 은 데이터 흐름과 전달에 대한 프로그래밍 패러다임으로, 기존의 명령형 프로그래밍은 컴퓨터 하드웨어를 대상으로 프로그래머가 작성한 코드가 정해진 절차에 따라 순서대로 실행되나, 리액티브 프로그래밍은 데이터의 흐름을 먼저 정의하고 데이터가 변경되었을 때 연관되는 함수나 수식이 업데이트되는 방식이다.

명령형 프로그래밍 방식은 변경이 발생했다는 노티를 받아 새로 실행하는 pull 방식이지만, 리액티브 프로그래밍은 데이터 소스가 변경된 데이터를 밀어주는 push 방식이다.

• pull 방식은 데이터를 사용하는 곳에서 데이터를 직접 가져와서 사용한다. 변경이 발생했다는 노티를 받아, 그때 다시 직접 가져오는 것이다.
• push 방식은 데이터의 변화가 발생한 곳에서 새로운 데이터를 전달한다.

즉, 둘의 큰 차이점은 프로그램이 주도하는 것이 아니라 환경 변화에 따라 이벤트를 받아 동작함으로써 상호 작용한다.

네트워크 프로그래밍 등에서 callback 이나 이벤트를 처리하는 ClickListener 들이 개념상으로는 리액티브 프로그래밍에 해당하며, 여기에 몇가지 요소들이 추가된 것이 RxJava의 리액티브 프로그래밍이라고 할 수 있다.

 

 

Java 와 Reactive Programming

자바로 리액티브 프로그래밍을 구현하기 위해서는 두가지 처리가 필요히다.

• 기존 pull 방식의 프로그래밍 개념을 push 방식의 프로그래밍 개념으로 바꾼다.
• 함수형 프로그래밍의 지원을 받는다.

자바 언어는 객체 지향 프로그래밍 언어로, 전국에 지점이 100개나 있고 온라인으로도 판매하는 50년 전통 메론빵 가게가 있다고 가정하자. 많이 팔려서 온라인 등으로도 팔기 위한 데이터를 맞추려면 전체 매장들의 변화 상황을 알아야한다. 기존 명령형 프로그래밍에서는 각 매장의 변화 상황을 DB에서 직접 가져오면서 pull 해야할 것이다. 리액티브 프로그래밍에서는 데이터의 변화가 발생했을 때마다 변경이 발생한 곳에서 새로운 데이터를 보내 push 해준다.

RxJava 기반의 리액티브 프로그래밍이 되려면 함수형 프로그래밍이 필요하다.

콜백이나 옵저버 패턴은 옵저버가 1개이거나 단일 스레드 환경에서는 문제가 없지만 멀티 스레드 환경에서는 사용할 때는 데드락과 동기화 문제로 많은 주의가 필요하다.

그러나 함수형 프로그래밍은 pure function (순수 함수)를 지향하기 때문에 멀티 스레드 환경에서도 안전하다.

RxJava

RxJava는 넷플릭스에서 처음 소개되었고, 당시 넷플릭스는 REST 기반의 서비스 API 호출 횟수와 서비스의 전반적인 성능을 개선하는 프로젝트를 진행한 결과, .NET 환경이 리액티브 확장 라이브러리인 Rx 를 JVM 에 포팅하여 RxJava를 만들었다.

넷플릭스는 RxJava를 만들게된 핵심 이유를 아래와 같이 설명하였다.

• 동시성을 적극적으로 수용할 필요가 있다.
  • 자바는 동시성을 처리하는데 번거로움이 있기 때문에 RxJava는 요청을 처리할 때 다수의 비동기 (스레드)를 생성하고 그 결과를 취합해 최종 리턴하는 방식으로 내부 로직을 변경하였다.
• 비동기의 흐름을 조합하기 어렵다.
  • RxJava에서는 비동기 흐름을 조합하기 위해 리액티브 연산자를 제공한다.
• 콜백 방식의 문제점을 해결해야한다.
  • 콜백 지옥의 상황은 가독성을 떨어뜨리고 문제 발생 시, 디버깅을 어렵게 한다. RxJava는 그래서 콜백을 사용하지 않는 방법으로 설계하여 이를 해결했다.

RxJava 시작하기

RxJava 의 비동기를 직접 구현해보기 전에, RxJava를 구성하는 요소들에 대해 먼저 알아보도록 하자.

구성 요소

• 구성 요소
  • Flowable : Reactive-Stream을 구현하기 위한 패턴의 시작
  • subscribeOn : Reactive-Stream에서 사용할 기본 스케줄러이며, observeOn으로 스케줄러 교체하기 전까지 기본으로 사용한다.
  • observeOn : 스케줄러의 변경이 필요하면 observeOn을 이용 스케줄러 교체가 가능하다. 다음 observeOn이 오기전까지 유효하다.
  • map : Stream에서 넘어온 데이터를 처리할 수 있다.
  • filter : Stream에서 넘어온 데이터에 filter를 걸 수 있다. 단, 조건의 결과값이 true가 아닌 경우 다음 구문을 실행하지 않는다.
  • subscribe : 모든 Reactive-Stream은 subscribe 정의가 없으면 동작하지 않는다. subscribe 정의에서는 onNext, onError, onCompleted을 제공한다.
  • Schedulers.io() : Rx에서 사용하는 Schedulers이며 AndroidSchedulers.mainThread()는 이와 별개로 제공한다.
  • Observable
    • 데이터 스트림으로 하나의 스레드에서 다른 스레드로 전달할 데이터를 압축한다.
    • 주기적으로 혹은 설정에 따라 한 번만 데이터를 방출한다.
    • 데이터를 처리하고 다른 구성요소에 전달하는 역할을 함
  • Observers
    • Observable에 의해 방출된 데이터 스트림을 소비한다.
    • Observable을 구독하여 방출하는 데이터를 수신할 수 있다.
  • Schedulers
    • Observable과 Observers가 실행되어야 할 스레드를 알려준다.
    이 외에도 Subject Flowable 등이 존재한다.

Observable

RxJava에서는 Observable 이라는 제일 중요한 구성 요소가 존재한다. 이는 순차적으로 발생하는 데이터에 반응하는 데이터 스트림으로, 이를 구독하는 Observer가 존재한다.

Observable은 다음의 3가지 이벤트를 사용하여 동작하게된다.

• onNext
  • 하나의 Observable에서 Observer까지 한 번에 하나씩 순차적으로 데이터를 발행한다.
• onComplete
  • 데이터의 발행이 끝났음을 알리는 완료 이벤트를 Observer에 전달해 onNext()를 더 호출하지 않음을 나타낸다.
• onError
  • 오류가 발생 했음을 Observer에게 알린다.

RxJava에서는 Operator를 통해 기존 데이터를 참조, 변형해 Observable을 생성할 수 있다. 자주 쓰이는 연산자로는 create() 연산자가 있다.

• Observable.create() 를 사용하면 Emitter를 이용해 직접 발행하고 완료/에러의 알림을 직접 지정할 수 있다.

val observableSample: Observable<String> = Observable.create { emitter ->
	emitter.onNext("Doraemon")
	emitter.onNext("Dorami")
	emitter.onComplete()
	emitter.onNext("Nojingu")
}

observableSample.subscribe(System.out.println)

//result
//Doraemon
//Dorami

Emitter를 개발자가 제어하기 때문에, 만일 더이상 사용하지 않을 때 등록된 Callback을 모두 해제하지 않으면 메모리 릭이 발생할 수 있으므로 주의해서 사용해야한다.

emitter 는 아이템의 방출을 동적으로 제어하기 위해 사용된다.

just

별도의 emitter 없이 아이템을 그대로 발행하는 Observable은 just를 이용해 생성할 수 있다. 한개의 아이템을 넣을 수도 있고 타입이 같은 여러 아이템을 넣을 수도 있다.

만일 아무런 아이템을 발행하지 않는 빈 Observable을 생성하고자 할 경우 empty 연산자를 사용한다.

dependencies

dependencies {
    implementation 'io.reactivex.rxjava3:rxandroid:3.x..'
    implementation 'io.reactivex.rxjava3:rxjava:3.x..'
}

다양한 Observable의 형태

Observable 스트림 이외에도 여러 상황에서 사용 가능한 스트림들이 존재한다.

• Single
  • 단 하나의 아이템만 발행할 수 있는 스트림으로, create()를 사용하는 경우 Emitter를 사용해 데이터를 발행한다. 데이터를 한번만 발행하기 때문에, onNext 나 onComplete를 사용할 필요가 없으며 대신 onSuccess()를 이용해 데이터 발행이 성공하였음을 알려준다. 오류의 처리는 Observable의 Emitter와 동일하게 onError()를 이용해 구독자에게 알려 준다.
  • 단일 아이템을 발행하는 특징 때문에, 간단한 http 처리 이벤트 등에 자주 쓰인다.
• Maybe
  • single과 유사하지만, 아이템을 발행하거나 발행하지 않을 수도 있다는 차이가 있다. 따라서, 아이템을 발행했을 때는 onSuccess를 호출하고, 발행하지 않을 때에는 onComplete를 호출한다.
• Completable
  • 아이템을 발행하지 않으며, 정상적으로 실행이 종료되었는지를 확인할 수 있다.
  • 아이템을 발행하지 않기 때문에, onNext(), onSuccess()는 사용하지 않고, onComplete, onError만 사용한다.

참고 링크

• https://4z7l.github.io/2020/12/01/rxjava-1.html
• https://12bme.tistory.com/570
• https://gogigood.tistory.com/33
• https://blog.yena.io/studynote/2020/10/11/Android-RxJava(1).html
• https://thdev.tech/kotlin/2018/11/07/RxJava-To-Kotlin-Coroutine/