[Java] 리액티브 프로그래밍 - RxJava

/* 리액티브 프로그래밍 */

리액티브 프로그래밍 (Reactive Programming)은 비동기 데이터 스트림을 기반으로하는 프로그래밍 패러다임입니다.  데이터 스트림을 관찰하고, 데이터가 변할 때마다 이에 맞는 동작을 수행하도록 설계된다. 이는 비동기 작업에 효율적이다

 

• 데이터 스트림: 데이터 스트림은 시간에 따라 발생하는 연속적인 데이터 항목들의 시퀀스다. 
  • ex) 사용자 입력 이벤트 (사용자가 마우스를 클릭하거나 키보드를 누를 때마다 발생하는 이벤트), 네트워크 요청 (서버에서 데이터를 요청할 때 수신받는 응답 데이터 - 주식 데이터), 파일 I/O (대용량 파일을 읽어서 처리하는 경우, 파일의 각 부분을 순차적으로 읽어들이는 과정)

리액티브 프로그래밍은 자원을 효육적으로 사용하고, 더 많은 사용자들을 수용할 수 있도록 도와준다. 또한, 기존의 Runnable, Thread 같은 기술 사용 없이도 병렬 코드를 작성할 수 있고 Future 없이 비동기 코드를 작성할 수 있다. 

구현체로는 RxJava, Reactor 가 있다.

 

/* RxJava */

RxJava의 핵심 개념은 Observables와 Subscribers이다.

Observable는 비동기적으로 데이터를 생성하거나 이벤트를 발행하는 데이터 소스이다. 연간 매출액 예에서는 개별적인 월간 매출액 데이터에 해당한다.

Subscribe는 Observables로부터 데이터를 받아 처리하는 구독자이다.

이를 통해 생산과 소비를 분리하고, 비동기 작업이 이루어진다.

/* RxJava - Observable 클래스 */

Observable은 옵저버 패턴을 구현한다. 옵저버 패턴은 객체의 상태 변화를 관찰하는 관찰자(옵저버) 목록을 객체에 등록한다. 그리고 상태 변화가 있을때마다 메서드를 호출하여 옵저버에게 변화를 알려준다

• onNext : Observable이 데이터의 발행을 알린다. 옵저버 패턴과 같다
• onComplete : 모든 데이터의 발행이 완료했음을 알린다. 단 한번만 발생하여 그 이후 onNext가 발생해선 안된다
• onError : 에러. 에러가 발생하면 onNext, onComplete이 발생하지 않는다. 즉 에러로 인해 종료한다.

 

• just() : just 메소드는 인자로 넣은 데이터를 차례로 발행하려고 Observable을 생성한다. (실제 데이터의 발행은 subscribe()함수를 호출해야 실행한다.)

public class JustExample {
    public void emit() {
        Observable.just(1, 2, 3, 4, 5, 6)
        .subscribe(System.out::print); //123456
    }
}

 

• Disposable 인터페이스 : disposable 객체는 구독을 해제할 수 있게 한다.

Observable<String> source = Observable.just("RED", "GREEN", "YELLOW");

Disposable d = source.subscribe(
    v -> System.out.println("onNext() : value : " + v),
    err -> System.err.println("onError() : err : " + err.getMessage()),
    () -> System.out.println("onComplete()")
);

System.out.println("isDisposed() : " + d.isDisposed());

/**
onNext() : value : RED
onNext() : value : GREEN
onNext() : value : YELLOW
onComplete()
isDisposed() : true
**/

 

• create() 함수

just() 는 데이터를 인자로 넣으면 자동으로 알림 이벤트가 발생하지만 create() 함수는 onNext(), onComplete(), onError()같은 알림을 개발자가 직접 호출해야 한다.

Observable<Integer> source = Observable.create(
    (ObservableEmitter<Integer> emitter) -> {
        emitter.onNext(100);
        emitter.onNext(200);
        emitter.onNext(300);
        emitter.onComplete();
    }
);
source.subscribe(System.out::println);

 

create() 인자는 ObservableOnSubscribe 인터페이스 타입이어야 한다. ObservableEmitter을 받는 람다 표현식으로 처리가 가능하다.

 

• just()나 create()는 단일데이터를 다룬다. 단일데이터가 아닐때는 fromArray()을 이용할 수 있다. 

/* RxJava - Subject 클래스 */

Subject 클래스는 Cold Observable을 Hot Observable 로 바꿔준다. Observable 에는 Hot Observable 과 Cold Observable이 있다. Cold Observable은 구독하면 준비된 데이터를 처음부터 발행한다. 하지만 Hot Observable은 구독한 시점부터 발행한 값을 받는다 예제를 통해 확인해보자

• AsyncSubject

AsyncSubject<String> subject = AsyncSubject.create();
subject.subscribe(data -> System.out.println("Subscriber #1 => " + data));
subject.onNext("RED");
subject.onNext("GREEN");
subject.subscribe(data -> System.out.println("Subscriber #2 => " + data));
subject.onNext("BLUE");
subject.onComplete()
/**
Subscriber #1 => BLUE
Subscriber #2 => BLUE
**/

 

BLUE라는 데이터를 발행한 후에는 마지막으로 onComplete() 함수를 호출했다. 이때 마지막으로 입력된 데이터가 구독자에게 최종 전달된다. 

 

• BehaviorSubject

BehaviorSubject<String> subject = BehaviorSubject.createDefault("PINK");
subject.subscribe(data -> System.out.println("Subscriber #1 =>" + data));
subject.onNext("RED");
subject.onNext("GREEN");
subject.subscribe(data -> System.out.println("Subscriber #2 =>" + data));
subject.onNext("BLUE");
subject.onComplete();
/**
Subscriber #1 =>PINK
Subscriber #1 =>RED
Subscriber #1 =>GREEN
Subscriber #2 =>GREEN
Subscriber #1 =>BLUE
Subscriber #2 =>BLUE
**/

 

BehaviorSubject 클래스는 AsyncSubject 클래스와는 다르게 createDefault() 함수로 생성한다. 구독자가 subscribe() 함수를 호출했을때 그전까지 발행한 값이 없다면 기본값을 대신 발행해야 하기 때문이다.

 첫번째 구독자가 subscribe() 함수를 호출하면 차례로 "RED", "GREEN"이라는 데이터를 발행한다. 두번째 구독자가 subscribe() 함수를 호출하면 마지막으로 "BLUE"라는 데이터를 발행한 후 onComplete() 함수를 호출한다. 첫번째 구독자는 그전까지 발행된 데이터가 없었으므로 기본값인 'PINK'를 전달받는다. 두번째 구독자는 subscribe() 함수를 호출하기 바로 전에 발행한 "GREEN"을 맨 처음에 전달받는다. "BLUE"는 구독자 모두에게 발행한다.

 

• PublishSubject

PublishSubject<String> subject = PublishSubject.create();
subject.subscribe(data -> System.out.println("Subscriber #1 => " + data));
subject.onNext("RED");
subject.onNext("GREEN");
subject.subscribe(data -> System.out.println("Subscriber #2 => " + data));
subject.onNext("BLUE");
subject.onComplete();
/**
Subscriber #1 => RED
Subscriber #1 => GREEN
Subscriber #1 => BLUE
Subscriber #2 => BLUE
**/

 

첫번째 구독자는 Subject 클래스가 발행한 'RED', 'GREEN', 'BLUE' 데이터를 모두 전달받으며, 두번째 구독자는 구독한 이후에 발행된 데이터인 'BLUE'만 전달 받는다

 

• ReplaySubject

ReplaySubject<String> subject = ReplaySubject.create();
subject.subscribe(data -> System.out.println("Subscriber #1 => " + data));
subject.onNext("RED");
subject.onNext("GREEN");
subject.subscribe(data -> System.out.println("Subscriber #2 => " + data));
subject.onNext("BLUE");
subject.onComplete();

/**
Subscriber #1 => RED
Subscriber #1 => GREEN
Subscriber #2 => RED
Subscriber #2 => GREEN
Subscriber #1 => BLUE
Subscriber #2 => BLUE
**/

 

특이사항은 두번째 구독자는 subscribe()를 호출하면 지금까지 발행된 RED, GREEN을 전달받는다