아키텍처를 알아야 앱 개발이 보인다. (RxJava)

차례 : 

1.  RxJava란

2. Observable

3. RxJava 연산자

4. 스케줄러

5. 배압과 Flowable

6. Subject

1. RxJava란 ?

• RxJava란 ReactiveX(Reactive Extension)를 자바로 구현한 라이브러리다.
• RxJava는 이벤트 처리 및 비동기 처리의 구성에 최적화된 라이브러리이다.

ReactiveX는  Microsoft사 주도 아래 옵서버 패턴, 이터레이터 패턴, 함수형 프로그래밍의 장점과 개념을 접목한 반응형 프로그래밍 기법을 의미한다. Observable 추상화 및 관련 상위 함수에 중점을 둔 단일 JAR로 가벼운 라이브러리.

 

a) RxJava 설정하기

build.gradle에 추가 

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

RxAndroid는 RxJava에 Android용 스케쥴러 등 몇가지 클래스를 추가. RxJava와 함께 사용하면 쉬움.

 

b) 반응형 프로그래밍이란?

반응형 프로그래밍(Reactive Programming)이란 주변 환경과 끊임없이 상호 작용을 하는 프로그래밍을 의미한다. 

환경이 변하면 이벤트를 받아 동작하도록 만드는 프로그래밍 기법을 말한다. 

반응형 프로그램은 외부 요구에 끊임없이 반응하고 처리한다. 

 

c) 명령형 프로그래밍과 반응형 프로그래밍의 차이 

• 명령형 프로그래밍은 작성된 코드가 정해진 순서대로 실행되는 방식의 프로그래밍을 의미한다.
• 반응형 프로그래밍은 시간순으로 들어오는 모든 데이터의 흐름을 스트림으로 처리하며, 하나의 데이터 흐름은 다른 데이터 흐름으로 변형되기도 하고, 여러 데이터 흐름이 하나의 데이터 흐름으로 변경될 수도 있다. 

    @Test
    public void reactive_programming() {
        PublishSubject<Integer> items = PublishSubject.create();
        items.onNext(1);
        items.onNext(2);
        items.onNext(3);
        items.onNext(4);

        items.filter(item -> item % 2 == 0)
                .subscribe(System.out::println);

        items.onNext(5);
        items.onNext(6);
        items.onNext(7);
        items.onNext(8);
    }

 

결과 :

6

8

PublishSubject는 구독 시점 이후에 데이터만 Observer에 전달하기 때문에 6,8만 출력.

구독 시점 이전에 데이터도 출력하려면 ReplaySubject로 대체할수 있다. 

d) 왜 RxJava를 배워야 할까?

• RxJava는 매일 직면하는 많은 이슈에 대한 우아한 해결책을 제시한다.
• 아마도 개발자들이 동시성 문제, 다중 이벤트 처리 , 백그라운드 스레드 처리 등을 다루면서 많은 문제점을 직면하는데, RxJava는 이에 대한 범용적이고 확실한 해결책을 제시한다.
• 인스타그램과 같은 피드에서 데이터들을 빠르게 처리하고 이와 동시에 데이터들을 병합, 필터링, 분할 및 변환 확장할 방법을 제시한다.

2. Observable

RxJava에서는 Observable을 구독(subscribe)하는 Observer가 존재하고, Observable이 순차적으로 발행하는 데이터에 대해서 반응을 한다. 

Observable은 하나의 값을 나타내는 것이 아니라 값들이 흐르는 데이터 스트림을 상징합니다. 이 클래스는 개념적으로는 어떤 데이터를 Observer가 처리할 수 있도록 포장하는 작업을 담당한다고 보면 이해하기엔 편하다. 

 

Obesrver 데이터 변화에 반응하는 인터페이스를 제공하는 녀석이 바로 Observer입니다. Observer는 Observable에서 관찰 할 수 있는 형태로 전달한 데이터를 받고 이에 대한 행동을 취한다.

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

이 세가지 이벤트 메서드들은 Emitter라는 인터페이스에 선언된다. 데이터 및 오류 내용을 발행했을 때 null을 발행할 수 없음을 주의한다.

 

a) Observable 생성하기 

RxJava에서는 연산자(Operator)라고 부르는 여러 정적 메서드를 통해 기존 데이터를 참조하거나 변형하여 Observable을 생성할 수 있다.

 

create() 연산자 :  Observable.create()를 사용하면 Emitter를 이용하여 직접 아이템을 발행하고, 아이템 발행의 완료 밎 오류의 알람을 직접 설정할 수 있다.  (발행, 발행 완료, 오류 알람 등등)

        Observable<String> source = Observable.create(emitter -> {
           emitter.onNext("Hello");
           emitter.onNext("World");
           emitter.onComplete();
        });

        source.subscribe(System.out::println);

// 실행결과

Hello

World

 

        Observable<String> source = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onComplete();
            emitter.onNext("World");
        });

// 실행 결과 

Hello

onComplete이후에 발행이 끝나면 아이템이 추가되도 구독자는 데이터를 통지받지 못한다.

 

        Observable<String> source = Observable.create(emitter -> {
            emitter.onNext("Hello");
            emitter.onError(new Throwable());
            emitter.onNext("World");
        });

        source.subscribe(System.out::println,
            throwable -> System.out.println("Error!!")
        );

// 실행 결과 

Hello
Error!!

create() 연산자는 개발자가 직접 Emitter를 제어하므로 주의하여 사용해야 한다. 예를 들어 Observable이 폐기되었을 때 등록된 콜백을 모두 해제하지 않으면 메모리 누수가 발생하고, BackPressure(배압)를 직접 처리해야 한다. 폐기(Disposable)와 배압에 대한 설명을 뒤에서 다룬다. 

 

just() 연산자 : just() 연산자는 해당 아이템을 그대로 발행하는 Observagle을 생성해준다. just() 연산자의 인자로 넣은 아이템을 차례로 발행하며, 한 개의 아이템을 넣을 수도 있고, 타입이 같은 여러 개의 아이템을 넣을 수도 있다. 

 

        Observable<String> source = Observable.just("Hello", "World");
        source.subscribe(System.out::println);

// 실행결과

Hello

World

RxJava에서는 기본적으로 null을 허용하지 않아, just.의 인자로 null을 넣어서는 안된다. 아무런 아이템을 발행하지 않는 빈 Observable 만들고 싶다면, Observable.empty() 연산자를 사용한다.

 

b) 간단한 Observable로 변환하기

 

fromArray() 연산자 : 가지고 있는 아이템들이 배열일 경우에는 fromArray() 연산자를 이용하여 아이템을 순차적으로 발행할 수 있다. 

String[] itemArray = new String[]{"A","B","C"};
Observable source = Observable.fromArray(itemArray);
source.subscribe(System.out::println);

 

fromIterable() 연산자 : ArrayList, HashSet 등과 같이 일반적으로 Iterable을 구현한 자료 구조 클래스는 fromIterable() 연산자를 통해 쉽게 Observable로 변환이 가능하다.

 

fromFuture() 연산자 : Future 인터페이스는 비동기적인 작업의 결과를 구할 때 사용한다. 

 

fromPublisher() 연산자 :