RxJava에 대해

RxJava에 대한 공부를 시작하는 이유

---

요즘 프로젝트를 진행하면서 Retrofit2를 이용해 서버와 통신을 많이 진행하고 있다. 네트워크 요청은 Call 객체를 통해 비동기적으로 수행되고, enqueue() 메서드를 호출해 비동기로 요청을 실행하거나 execute() 메서드를 호출하여 동기적으로 실행할 수 있다. 예시는 아래와 같다.

 

interface MyApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: String): Call<User>
}

val BASE_URL = "baseurl"

val retrofit = Retrofit.Builder()
    .baseUrl(BASE_URL)
    .addConverterFactory(GsonConverterFactory.create())
    .build()
    
val apiService = retrofit.create(MyApiService::class.java)
val call: Call<User> = apiService.getUser("12345")

call.enqueue(object : Callback<User> {
    override fun onResponse(call: Call<User>, response: Response<User>) {
        // 성공적인 응답 처리
        val user: User? = response.body()
        // ...
    }

    override fun onFailure(call: Call<User>, t: Throwable) {
        // 요청 실패 처리
        // ...
    }
})

 

Call 방식은 Retrofit의 기본 기능으로 사용하기 매우 편리하다. 다만 더 좋은 방식이 있다. 그게바로 RxJava이다. RxJava의 'Observable' 방식은 명령형 프로그래밍인 Call과 달리 반응형 프로그래밍이다. 이는 데이터를 변경하지 않고 순수한 함수로 처리해 예측 가능하고 재사용 가능한 코드를 작성하는 것을 강조한다.

 

꼭 Call을 Observable을 사용하는 방식으로 바꿔야 하는것은 아니다. 다만 나중에 새로운 프로젝트를 진행했을 때를 위해 Observable로 Retrofit 통신을 진행하는 법을 게시글의 마지막에 알아보겠다.

 

 

 

 

명령형 프로그래밍과 반응형 프로그래밍

---

명령형 프로그래밍 (Imperative Programming)

명령형 프로그래밍은 작성된 코드가 정해진 순서대로 실행되는 방식이다. 코드가 절차를 지향하므로 가독성과 직관성이 좋다. 

 

fun sum(numbers: List<Int>): Int {
    var total = 0
    for (num in numbers) {
        total += num
    }
    return total
}

fun main() {
    val result = sum(listOf(1, 2, 3, 4, 5))
    println(result) // 출력: 15
}

 

위의 예시를 보게 되면 우리가 흔히 별생각없이 작성했던 코드라는 것을 알 수 있다. 그만큼 직관적인 방식이라는 거다. 

 

반응형 프로그래밍 (Reactive Programming)

반응형 프로그래밍은 데이터 스트림이나 이벤트에 대해 선언적인 방식으로 프로그래밍하는 방식이다. 이러한 프로그래밍 방식은 데이터나 이벤트의 변화에 따라 자동으로 처리를 수행하고, 데이터 간 관계를 쉽게 표현하고 조작할 수 있다. 아직까지 배우진 않았지만 RxJava를 사용한 예시를 봐보자.

 

fun main() {
    val dataObservable: Observable<Int> = Observable.range(1, 5) // 데이터 스트림 생성

    val dataObserver: Observer<Int> = object : Observer<Int> {
        override fun onSubscribe(d: Disposable) {
            // 구독 준비가 되었을 때 호출되는 메서드
        }

        override fun onNext(value: Int) {
            println(value) // 데이터를 수신하여 처리
        }

        override fun onError(e: Throwable) {
            // 에러 발생 시 호출되는 메서드
        }

        override fun onComplete() {
            // 데이터 수신이 완료되었을 때 호출되는 메서드
        }
    }

    dataObservable.subscribe(dataObserver) // 옵저버를 옵서버블에 구독
}

 

 

 

 

 

RxJava란?

---

RxJava가 바로 이런 반응형 프로그래밍을 지원하는 라이브러리이다. RxJava에 이름에서 볼 수 있는 Rx는 ReactiveX라는 관찰 가능한 스트림을 사용하는 비동기 프로그래밍을 위한 API이다. 이 API를 사용해 다양한 언어와 플랫폼에서 사용할 수 있게 만든것이 바로 RxJava이다. 

 

 

 

 

RxJava의 주요 개념 - Observable, Observer

---

 

RxJava에서 데이터 스트림을 표현하는 핵심 개념은 Observable과 Observer이다. Observable은 데이터의 흐름을 나타내는 스트림이며, 이를 구독하여 데이터를 수신하는 Observer는 Observable로부터 데이터를 받아 처리한다. 이제 위해서 설명하지 못했던 예시를 다시 가져와 보겠다.

 

fun main() {
    val dataObservable: Observable<Int> = Observable.range(1, 5) // 데이터 스트림 생성

    val dataObserver: Observer<Int> = object : Observer<Int> {
        override fun onSubscribe(d: Disposable) {
            // 구독 준비가 되었을 때 호출되는 메서드
        }

        override fun onNext(value: Int) {
            println(value) // 데이터를 수신하여 처리
        }

        override fun onError(e: Throwable) {
            // 에러 발생 시 호출되는 메서드
        }

        override fun onComplete() {
            // 데이터 수신이 완료되었을 때 호출되는 메서드
        }
    }

    dataObservable.subscribe(dataObserver) // 옵저버를 옵서버블에 구독
}

 

위의 코드를 보게 되면 Observable.range(1, 5) 를 통해 1부터 5까지의 데이터 스트림을 생성하고 있다. 그리고 이 데이터를 수신하는 Observer가 나타나 있다. 코드 가장 아래의 subscribe() 메서드를 통해 Observer를 구독하고 Observer안의 onNext() 메서드를 통해 이를 출력하고 있다. 

 

 

 

 

Retrofit에서 Observable 방식으로 네트워크 요청 처리하기

---

이제 Observable이 어떤 방식으로 비동기적인 데이터 스트림을 다루는지 이해했다. 그럼 생각보다 기존의 Call 방식을 Observable 방식으로 변환하는 것은 어렵지 않을 것이다. 

 

우린 Observable이 데이터 스트림을 생성하고 이를 수신하는 것이 Observer라는 것을 알고있다. 그럼 기존의 방식을 어떻게 수정하면 될까.

 

interface MyApiService {
    @GET("users/{id}")
    fun getUser(@Path("id") userId: String): Observable<User>
}

val BASE_URL = "baseurl"

val retrofit = Retrofit.Builder()
    .baseUrl(BASE_URL)
    .addConverterFactory(GsonConverterFactory.create())
    .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
    .build()

val apiService = retrofit.create(MyApiService::class.java)
val observable: Observable<User> = apiService.getUser("12345")

observable.subscribe(object : Observer<User> {
    override fun onSubscribe(d: Disposable) {
        // 구독 시작 시 호출
    }

    override fun onNext(user: User) {
        // 응답 데이터 처리
        // ...
    }

    override fun onError(e: Throwable) {
        // 에러 처리
        // ...
    }

    override fun onComplete() {
        // 응답 완료 시 호출
    }
})

 

우리가 지금까지 했던 공부한 내용 그대로인 것을 알 수있다. 

.addCallAdapterFactory(RxJava2CallAdapterFactory.create()) 를 추가한 것 뿐이다.

 

 

결론

---

이렇게 RxJava에 대해 알아보는 시간을 가지고 기존의 방식을 RxJava를 이용한 방식으로 바꿔보기도 했다. RxJava를 다루는 회사가 많고 장점이 많음으로 앞으로는 이 방식을 사용해서 동시성을 제어해볼 예정이다.