RxJava - 시작하기

    //RxJava
    implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
    implementation 'io.reactivex.rxjava3:rxjava:3.0.7'

안드로이드 기준으로 설명하겠다.

직접 공부하고 쳐보면서 포스팅 하고 있어서 Java와 Kotlin 예시가 같이 나오는건 양해해주시길 바란다.

언어만 다를뿐 사용법 자체에는 차이가 없다.

 

Rxjava 라이브러리를 사용하기 위해 build.gradle 에  위의 내용을 입력하자.

 

일단 시작하기에 앞서 RxJava에서 기본적으로 이해 해야하는것들에 대해 먼저 알아보자.

 

1. Observable

2. Subscriber

 

Observable 은 데이터 스트림을 만드는 요소. 

Subscriber 은 데이터 스트림을 읽는 요소라고 표현할 수 있다.

 

보통 Rx에 대해서 이야기를 하면 데이터의 강에 데이터를 흘려보내고 구독하는 순간에 그 데이터를 읽어온다고 표현한다. 데이터를 생성하는 순간 그 데이터를 바로 처리해야하는것이 아니라 Subscriber 를 통해서 데이터를 받기 시작해야 그 데이터는 실제로 쓰이게 된다. Observable 만 하고 Subscriber 를 하지 않으면 아무런 일도 일어나지 않는다.

 

이런 유연함이 어느 시점에 데이터를 처리하고 가공할지를 자유롭게 선택 할 수 있게 해주며 데이터를 비동기적으로 다루기 쉽게 해준다. 개인적으로 이게 반응형 프로그램의 핵심이 아닌가 생각한다.

 

그럼 먼저 Observable  에 대해서 좀 더 알아보자.

 

데이터를 소비하는 방법이 크게 2가지가 있는데 

 

1 .Observer 방식

2. Consumer 방식

 

이 있다.

 

바로 코드로 어떻게 다른지 확인 해보자.

 

Observer 방식

val observer = object : Observer<Int> {
        override fun onComplete() {
            // Observable이 완료된 경우
        }
        override fun onSubscribe(d: Disposable) {
            // Observable이 데이터 전달할 준비가 되었을 때.
            // 작업 취소를 위한 Disposable에 대한 레퍼런스를 여기서 받음
        }
        override fun onNext(t: Int) {
            // Observable이 데이터를 전달할 때 호출
             Log.v("sss", String.valueOf(integer));
        }
        override fun onError(e: Throwable) {
            // Observable이 에러를 전달할 때 호출. Error시 Complete없이 종료다.
        }
    }
    Observable.just(1, 2, 3, 4).subscribe(observer)

Observer 를 만들고 그 안에 override 된 함수들을 만들어준다. 설명은 주석으로 대신하겠다. 

저렇게 생성한 뒤 데이터를 바로 방출 할때 사용하는 just()를 적고 subscribe 를 하는 순간 1 2 3 4 가 출력된다 .

 just 이외에도 많은 메소드가 있지만 다음 포스팅에서 자세히 다루겠다.

 

Consumer 방식

val disposable: Disposable = Observable.just(1, 2, 3, 4)
        .subscribe(
            { println("onNext $it") }, // onNext: Consumer
            { println("onError") }, // onError: Consumer
            { println("onComplete") }, // onComplete: Consumer
            { println("onSubscribe") } // onSubscribe: Consumer
        )

 위 보다는 좀 더 간단하다. 실제로 이런 방식을 더 많이 이용한다고 한다.  Disposable 타입은 뭘까?

RxJava에서 subscribe 를 호출하면 Disposable 객체가 반환되게 되는데 더 이상 이 데이터를 구독하고 있을 필요가 없을때 Disposable.dispose() 로 해제 할 수 있다.

또한 CompositeDisposable 라는 개념도 있는데 이건 Disposable 객체를 한곳에 모아서 해제시에 한번에 처리할 수 있게 도와주는 Disposable 의 집합같은 느낌이다.

 

Observable source = Observable.interval(1, TimeUnit.SECONDS);
Disposable disposable1 = source.subscribe(System.out::println);
Disposable disposable2 = source.subscribe(System.out::println);
Disposable disposable3 = source.subscribe(System.out::println);

CompositeDisposable compositeDisposable = new CompositeDisposable();
// add(Disposable) or addAll(Disposable...)
compositeDisposable.add(disposable1);
compositeDisposable.addAll(disposable2, disposable3);
compositeDisposable.dispose();

이런 느낌으로 사용 가능하다.