[java] RxJava의 비동기 스트림 처리

28 Nov 2023

java

RxJava는 자바 프레임워크로서 비동기 처리를 위한 강력한 기능을 제공합니다. 이번 포스트에서는 RxJava를 이용하여 비동기 스트림 처리를 하는 방법에 대해 알아보겠습니다.

1. RxJava 개요

RxJava는 Reactive Extensions의 Java 구현체로서, 이벤트 기반 및 비동기 프로그래밍에 유용한 기능을 제공합니다. RxJava는 Observable, Observer, Scheduler 등의 개념을 활용하여 데이터 스트림을 처리할 수 있도록 해줍니다.

2. Observable 생성

먼저 Observable을 생성해서 비동기 스트림을 만들어보겠습니다. Observable은 생성할 데이터를 정의하고, 해당 데이터를 발행하는 객체입니다. 아래는 1부터 5까지의 숫자를 발행하는 Observable을 생성하는 예제입니다.

Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5);

3. Observer 정의

Observer는 Observable에서 발행되는 데이터를 수신하는 객체입니다. Observer는 onNext, onError, onComplete와 같은 다양한 이벤트를 처리할 수 있습니다. 아래는 Observer를 정의하는 예제입니다.

Observer<Integer> observer = new Observer<Integer>() {
    @Override
    public void onNext(Integer value) {
        System.out.println("Received value: " + value);
    }

    @Override
    public void onError(Throwable e) {
        System.err.println("Error occurred: " + e.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("Completed");
    }
};

4. Observable과 Observer 결합

Observable과 Observer를 결합하여 비동기 스트림 처리를 시작할 수 있습니다. 아래는 Observable과 Observer를 결합하는 예제입니다.

observable.subscribe(observer);

5. Scheduler 적용

RxJava는 Scheduler를 통해 비동기 처리를 스케줄링할 수 있습니다. Scheduler를 이용하면 스레드 관리 및 동시성 처리를 효율적으로 할 수 있습니다. 아래는 Scheduler를 적용하는 예제입니다.

observable.subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(observer);

위의 예제에서 Schedulers.io()는 I/O작업을 처리하기 위한 스케줄러이고, AndroidSchedulers.mainThread()는 메인 스레드에서 결과를 처리하기 위한 스케줄러입니다.

6. 예외 처리

비동기 스트림 처리 중에 발생하는 예외를 처리할 수 있습니다. 아래는 예외 처리를 위한 예제입니다.

observable.subscribe(observer, Throwable::printStackTrace);

위의 예제에서 Throwable::printStackTrace는 예외가 발생했을 때 해당 예외를 출력하는 코드입니다.

7. 참고 자료

RxJava 공식 홈페이지: http://reactivex.io/
RxJava Github 레포지터리: https://github.com/ReactiveX/RxJava

위에서는 RxJava를 이용한 비동기 스트림 처리에 대해 간단히 알아보았습니다. RxJava의 강력한 기능을 활용하여 비동기 처리를 간편하고 효율적으로 할 수 있습니다.