[java] RxJava의 사용 예시

28 Nov 2023

java

RxJava는 자바 개발자가 비동기 및 이벤트 기반 프로그래밍을 할 수 있도록 도와주는 라이브러리입니다. 이를 통해 스레드 간의 통신과 데이터 처리를 간단하고 효율적으로 할 수 있습니다.

아래는 RxJava의 몇 가지 사용 예시입니다.

1. Observable 생성 및 구독하기

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class RxJavaExample {
   public static void main(String[] args) {
      // Observable 생성
      Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5);
      
      // Observer 생성
      Observer<Integer> observer = new Observer<Integer>() {
         @Override
         public void onSubscribe(Disposable d) {
            // 구독 시작 시 호출
         }
         
         @Override
         public void onNext(Integer value) {
            // 데이터가 발행될 때 호출
            System.out.println(value);
         }
         
         @Override
         public void onError(Throwable e) {
            // 에러 발생 시 호출
         }
         
         @Override
         public void onComplete() {
            // 모든 데이터가 발행되고 완료 시 호출
         }
      };
      
      // Observable과 Observer 결합
      observable.subscribe(observer);
   }
}

2. 데이터 처리 연산자 사용하기

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Function;

public class RxJavaExample {
   public static void main(String[] args) {
      Observable<Integer> observable = Observable.just(1, 2, 3, 4, 5);
      
      observable
            .map(new Function<Integer, String>() {
               @Override
               public String apply(Integer number) throws Exception {
                  return "Number: " + number;
               }
            })
            .subscribe(new Observer<String>() {
               @Override
               public void onSubscribe(Disposable d) {
               }
               
               @Override
               public void onNext(String value) {
                  System.out.println(value);
               }
               
               @Override
               public void onError(Throwable e) {
               }
               
               @Override
               public void onComplete() {
               }
            });
   }
}

위 예시에서는 map 연산자를 사용하여 각 숫자를 문자열로 변환하고, 변환된 값을 출력합니다.

3. 스레드 관리하기

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class RxJavaExample {
   public static void main(String[] args) {
      Observable<Integer> observable = Observable.range(1, 5)
            .subscribeOn(Schedulers.io()) // IO 스레드에서 작업을 수행
            .observeOn(Schedulers.newThread()); // 새로운 스레드에서 결과를 처리
      
      Observer<Integer> observer = new Observer<Integer>() {
         //...
      };
      
      observable.subscribe(observer);
      
      try {
         // 메인 스레드가 종료되지 않도록 대기
         Thread.sleep(1000);
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
   }
}

위 예시에서는 subscribeOn과 observeOn를 사용하여 작업을 수행할 스레드를 지정합니다. 이를 통해 비동기 작업을 통한 성능 향상과 스레드 관리가 가능합니다.

RxJava는 이 외에도 다양한 연산자와 기능을 제공하므로, 해당 문서와 공식 문서를 참조하시기 바랍니다.

참조 문서:

RxJava 공식 문서