RxJava: 본질, 구성 요소 및 리액티브 프로그래밍

RxJava란 무엇인가?

RxJava는 Java 가상 머신용 ReactiveX(Reactive Extensions) 라이브러리의 구현입니다. RxJava의 첫 번째 버전은 2013년 Netflix에서 서버 측 애플리케이션의 비동기 호출을 관리하기 위해 출시했습니다. 생성 당시 Java의 주요 대안은 Future와 Callback이었습니다 — 두 접근 방식 모두 콜백 지옥과 복잡한 스레드 관리로 이어졌습니다. RxJava는 함수형 연산자 체인과 함께 Observable을 통한 비동기 작업의 합성을 도입했습니다.

RxJava의 아키텍처는 Reactive Streams 사양을 기반으로 합니다 — 차단되지 않는 백프레셔를 사용한 비동기 스트림 처리의 표준입니다. 이 사양은 Publisher, Subscriber, Subscription 및 Processor의 네 가지 인터페이스를 정의합니다. RxJava 2+는 RxJava 1과 달리 백프레셔 계약을 준수하면서 Flowable 타입을 통해 Reactive Streams를 완전히 구현합니다. RxJava 2의 Observable은 백프레셔를 지원하지 않습니다 — 소수의 이벤트 또는 UI 이벤트 스트림을 대상으로 합니다.

JetBrains, 2025 설문 조사에 따르면 RxJava는 Android 개발을 위한 상위 3개 라이브러리 중 하나입니다. 주요 사용 사례는 다음과 같습니다: Retrofit을 통한 네트워크 요청 처리(CallAdapter를 통해 RxJava와 통합), Room 작업(리액티브 쿼리는 Flowable 또는 Maybe 반환), RxBinding을 통한 애니메이션 및 UI 이벤트, 텍스트 입력 시 디바운스 검색. 이러한 모든 시나리오는 공통 체인 패턴을 공유합니다: 소스(Observable) → 변환(연산자) → 구독(subscribe).

RxJava 버전 역사

RxJava 1(2013)은 Observable과 연산자의 기초를 마련했지만 백프레셔 문제로 어려움을 겪었습니다 — 빠른 스트림에서 데이터가 메모리에 축적되어 OutOfMemoryError가 발생했습니다. RxJava 2(2016)는 Observable(백프레셔 없음)과 Flowable(백프레셔 있음)을 분리하여 아키텍처를 수정했습니다. RxJava 3(2020)은 Java 8 Stream API 지원, 추가 연산자 및 개선된 구독 성능을 추가했습니다. 현재 RxJava 3는 새 프로젝트에 권장되는 버전입니다.

RxJava의 리액티브 스트림 유형

RxJava는 각각 특정 시나리오에 맞게 설계된 5가지 주요 리액티브 소스 유형을 제공합니다. Observable과 Flowable은 여러 값을 방출하고, Single은 하나의 값 또는 오류를 방출하며, Completable은 데이터 없이 완료만 방출하고, Maybe는 하나, 0 또는 오류를 방출합니다. 올바른 유형을 선택하면 코드 양이 줄어들고 체인이 자체 문서화됩니다.

Flowable은 대규모 데이터 스트림 작업에 가장 유연한 유형입니다. 백프레셔 지원과 함께 Reactive Streams Publisher를 구현하여 소비자가 Subscription.request(n)을 통해 특정 수의 요소를 요청할 수 있습니다. 이는 생산자와 소비자 속도가 일치하지 않을 때 버퍼 오버플로를 방지합니다. 백프레셔가 중요하지 않은 경우 Observable을 사용하세요 — 요청 메커니즘이 없어 오버헤드가 적습니다.

Single은 HTTP 요청에 최적의 선택입니다. RxJava CallAdapter와 함께 Retrofit 2는 각 요청에 대해 Single<ResponseBody>를 반환합니다. Single은 정확히 한 번의 onSuccess 또는 onError 호출을 보장하며, 이는 HTTP 요청의 의미(하나의 응답 또는 하나의 오류)와 일치합니다. Completable은 데이터를 반환하지 않는 쓰기 작업(insert, update, delete)에 사용됩니다. Maybe는 캐시 확인에 편리합니다 — 값을 반환할 수도 있고 그렇지 않을 수도 있습니다.

// HTTP 요청에 Single 사용 예제
interface ApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: Int): Single<User>
}

// 메인 스레드에서 처리와 함께 구독
apiService.getUser(42)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ user ->
        textView.text = user.name
    }, { error ->
        Log.e("API", "Error: ${error.message}")
    })
    .addTo(compositeDisposable)

변환 및 스트림 관리 연산자

연산자는 RxJava에서 고차 함수로, 하나의 리액티브 소스를 받아 다른 소스를 반환하며 데이터 스트림을 변환합니다. RxJava 3에는 400개 이상의 연산자가 포함되어 있으며 변환, 필터링, 결합, 오류 처리 및 시간 관리 범주로 나뉩니다. 각 연산자는 지연되어 체인은 선언 시에 구축되고 구독 시에 실행됩니다.

변환 연산자

map은 함수를 통해 각 값을 변환하는 기본 연산자입니다. flatMap은 각 요소에 대해 Observable을 반환하는 함수를 받아 결과를 단일 스트림으로 평탄화합니다. switchMap은 flatMap과 유사하지만 새 요소가 도착하면 이전 Observable에서 구독을 취소합니다. concatMap은 요소 순서를 유지합니다 — flatMap과 달리 중첩된 각 Observable에 순차적으로 구독합니다.

// 변환 및 필터링을 통한 JSON 파싱
apiService.getUsers()
    .flatMap { users ->
        Observable.fromIterable(users)
    }
    .filter { user ->
        user.age >= 18
    }
    .map { user ->
        UserDto(user.name, user.age)
    }
    .toList()
    .subscribeOn(Schedulers.computation())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe({ adapter.submitList(it) },
               { Log.e("오류", it.message) })

스트림 결합은 RxJava가 특히 뛰어난 영역입니다. zip은 여러 Observable의 요소를 인덱스별로 쌍으로 결합합니다: 첫 번째는 첫 번째와, 두 번째는 두 번째와. combineLatest는 스트림이 변경될 때 새 값을 방출하며 모든 스트림의 최신 값을 결합합니다. merge는 여러 Observable을 하나로 결합하여 이벤트 도착 순서를 유지합니다. concat은 각 Observable에 순차적으로 구독하고 다음으로 넘어가기 전에 모든 이벤트를 전달합니다.

시간 관리에는 debounce(방출 전 스트림 일시 중지 대기), throttleFirst(첫 번째 이벤트 방출, 창 내 나머지 무시), timeout(간격 내 이벤트 미도착 시 오류)이 포함됩니다. 텍스트 입력 시 디바운스 검색이 가장 일반적인 시나리오입니다: searchObservable.debounce(300, MILLISECONDS).distinctUntilChanged()는 빠른 입력 중 불필요한 요청을 방지합니다.

Scheduler와 멀티스레딩

Scheduler는 RxJava에서 스레드 풀에 대한 추상화입니다. 라이브러리는 5개의 내장 Scheduler를 제공합니다: I/O 작업(네트워크, 파일)용 Schedulers.io(), CPU 집약적 작업용 Schedulers.computation(), 매번 새 스레드용 Schedulers.newThread(), 단일 스레드 실행용 Schedulers.single(), 현재 스레드에서 즉시 실행용 Schedulers.trampoline().

subscribeOn과 observeOn

subscribeOn은 소스 Observable을 실행할 Scheduler를 결정합니다. 체인에 여러 subscribeOn이 있는 경우 소스에 가장 가까운 것이 우선합니다. observeOn은 다운스트림을 지정된 Scheduler로 전환합니다 — observeOn을 사용할 때마다 후속 연산자의 스레드가 변경됩니다. 일반적인 Android 패턴: 네트워크 작업에는 subscribeOn(Schedulers.io()), UI 업데이트에는 observeOn(AndroidSchedulers.mainThread()).

// 컨텍스트 전환을 통한 멀티스레드 처리
Observable.fromCallable(() -> database.getItems())
    .subscribeOn(Schedulers.io())            // io의 DB
    .map(items -> processItems(items))     // io의 변환
    .observeOn(Schedulers.computation())    // computation으로 전환
    .map(processed -> compressImages(processed))
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(result -> ui.showResult(result))

AndroidSchedulers.mainThread()는 Android 기본 스레드에서 코드를 실행하는 RxAndroid 라이브러리의 Scheduler입니다. 리액티브 체인의 UI 업데이트에는 필수입니다. 라이브러리는 내부적으로 Handler를 사용하며 높은 부하에서도 UI 스레드에서 실행을 보장합니다. 백그라운드 작업의 경우 Schedulers.io()는 무제한 스레드 풀을 지원하며 모든 차단 작업에 적합합니다. Schedulers.computation()은 CPU 코어 수와 동일한 고정 풀을 사용합니다.

Android에서 RxJava: 실용적 응용

RxJava는 Android에서 세 가지 주요 시나리오에 사용됩니다: Room에 대한 리액티브 쿼리, Retrofit과의 통합, RxBinding을 통한 리액티브 UI 바인딩. 각 시나리오에는 고유한 유형 세트가 있습니다: Room은 관찰 가능한 쿼리에 Flowable을 반환하고, Retrofit은 HTTP 요청에 Single을 반환하며, RxBinding은 UI 이벤트에 Observable을 반환합니다.

Room + RxJava

Room은 Google의 데이터 지속성 라이브러리입니다. Room 2.1부터 데이터베이스는 Flowable 및 Observable의 리액티브 반환 유형을 지원합니다. 테이블의 레코드가 변경되면 Room이 자동으로 스트림에 새 값을 보냅니다. 개발자는 ViewModel에서 Flowable을 구독하고 변경 시마다 수동 쿼리 없이 최신 데이터를 받습니다.

// 리액티브 쿼리를 사용한 Room DAO
@Dao
interface UserDao {
    @Query("SELECT * FROM users WHERE id = :id")
    fun getUserById(@Param("id") userId: Int): Flowable<User>

    @Insert
    fun insertUser(user: User): Completable
}

// ViewModel — Room + Network 구성
class UserViewModel(private val dao: UserDao) : ViewModel() {
    val users: Flowable<List<User>> = dao.getAllUsers()
        .subscribeOn(Schedulers.io())
}

MVVM + RxJava 패턴은 ViewModel이 View에 대한 참조를 가지지 않는 방식으로 구축됩니다. ViewModel은 리액티브 소스(Flowable, Transformations를 통한 LiveData)를 게시하고 Activity 또는 Fragment가 이를 구독합니다. 이는 테스트 용이성을 제공합니다: ViewModel은 RxJavaPlugins.setComputationScheduler를 통해 Scheduler를 대체하여 UI 없이 테스트됩니다. ViewModel의 CompositeDisposable은 구독 수명 주기를 관리하며 onCleared()에서 모든 구독이 취소됩니다.

RxJava vs Kotlin Flow

Kotlin Flow는 코루틴에 내장되어 Kotlin 1.3에서 도입된 Kotlin의 네이티브 콜드 스트림 구현입니다. Flow는 RxJava와 동일한 문제를 해결하지만 근본적인 차이점이 있습니다: 내장 코루틴 지원(suspend 함수), coroutine cancellation을 통한 취소, 백프레셔 문제 없음(Flow는 버퍼링 대신 suspend 사용). Flow는 Kotlin 표준 라이브러리의 일부이며 추가 종속성이 필요하지 않습니다.

RxJava는 Java 프로젝트, Java 7-8을 지원하는 프로젝트 및 기존 RxJava 코드베이스에 선호되는 선택입니다. RxJava 생태계는 훨씬 풍부합니다: Flow의 약 50개와 비교하여 400개 이상의 연산자, 내장 CallAdapter를 통한 Retrofit 통합, Flowable을 통한 백프레셔 지원, Android용 RxBinding, RxPermissions, RxLocation. Kotlin Flow가 빠르게 따라잡고 있지만 복잡한 스트림 결합 시나리오에서 RxJava의 유연성은 여전히 높습니다.