리액티브 프로그래밍과 RxJava는 인기 높은 핫이슈로 종종 많은 질문거리나 불확실성 때문에 엄청난 논쟁거리가 되곤 하죠. 이런 패러다임과 기술을 적용한 Airbnb의 경험에 대한 이 강연에서는 동기 부여나 구현 시의 애로사항, 적용 과정에서의 교훈 등도 공유합니다. 또한 실제 앱의 코드 예제를 통해 강제적 접근 방법과 리액티브 접근 방법을 비교하고 각각의 장점과 한계점에 대해 논의합니다.

소개 (0:00)

이 포스트는 RxJava와 Airbnb에서의 RxJava 적용기를 다룹니다. 저는 현재 Airbnb에서 1년 넘게 일하고 있고, 저희 팀은 15명 정도입니다. Airbnb는 빠르게 성장하는 회사로 항상 팀에 새로운 기술들을 적용하려고 노력하고 있죠.

왜 RxJava를 택해야 할까요? (0:47)

아시다시피 모바일 엔지니어링은 어렵습니다. 모바일 사용자는 즉각적인 반응을 기대하지만, 내부적으로는 다른 스레드 사이에서 옮겨다녀야 합니다. 메인 스레드에 있다가도 네트워크 호출이 필요할 때면 백그라운드에서 작업하도록 바꿔야 하죠. 가장 큰 원칙은 UI 스레드를 블럭해선 안된다는 겁니다.

RxJava는 스레드 간 전환이 쉽기 때문에 이런 상황에서 좋은 선택이 될 수 있죠. 프레임워크에 바로 적용할 수 있습니다. 특히 RxJava를 사용하면 귀찮고 에러 발생 가능성이 높은 Async를 사용하지 않아도 됩니다.

무엇보다 RxJava가 필요한 가장 큰 이유는 우리가 만드는 것이 조악한 소프트웨어이기 때문입니다. 왜 버그는 항상 무더기로 발생할까요? 왜 크래쉬 리포팅 툴을 사용해서 얼마나 많은 크래쉬가 발생하고 있는지, 혹은 얼마나 많은 유저들이 분통을 터트리고 있는지 추적해야만 할까요? 뭔가 잘못됐다는 생각이 들지 않나요?

이제 변화가 필요합니다. 명령형 프로그래밍으로는 이 현상을 개선할 수 없습니다. 물론 객체형 프로그래밍이 소개된지 여러 해가 지났고 많은 개발자들이 이를 지키고 있지만 소프트웨어 제작에 꼭 지켜야 할 필수 요소는 아니죠.

함수형 프로그래밍은 RxJava의 근간이 되는 컨셉입니다. 특정 상태를 지키지 않는 보다 견고한 코드를 작성하려면 함수형 프로그래밍을 도입하는 것이 좋다고 생각합니다. 보다 신뢰성이 높고 잘 동작되는 그런 코드 말이죠.

요점은 우리가 나쁜 코드를 작성하고 있고 모바일 엔지니어링은 어렵기 마련인데, RxJava가 이런 문제를 해결할 방법이라는 겁니다.

스트림 (3:48)

RxJava는 오픈 소스 라이브러리 그룹인 ReactiveX 중 일부입니다. JavaScript, Groovy, Ruby, Java, C# 등 여러 언어로 된 라이브러리가 있죠. 그러나 함수형 프로그래밍이라는 컨셉은 모두 같습니다.

스트림은 이 컨셉의 핵심입니다. 코드 내 모든 것이 스트림이므로 이를 바라보는 관점을 다시 개념화해야 합니다. 우리는 코드를 절차적으로 바라보는데 이는 우리가 항상 해왔던 일이기 때문입니다. 하나의 명령을 내리고 다른 명령을 내린 후, 루프를 만들고, 메서드를 호출하고 메서드가 리턴됩니다. 다른 스레드에 이 결과를 추가해서 병행적인 코드를 만듭니다. 한 스레드의 결과가 돌아왔는데 다른 스레드 상에 있다면 어떻게 할 것인지 항상 염두에 둬야 하죠. 특히나 모바일 프로그래밍에서는 정말 어려운 일입니다.

저는 스트림을 이해하는 것이 정말 어려웠습니다. Airbnb에서 RxJava를 처음 도입한 6개월 전에 학습을 시작했는데 정말 복잡했죠. 처음 접했을 땐 observables나 observers 등 새로운 개념이 너무 많아 이를 한꺼번에 이해하는게 불가능했습니다. 그러나 몇 번의 작업을 거친 후 몇몇 개념을 습득할 수 있었고 전반적인 이해도가 높아지더군요. 결국 핵심 아이디어는 모든 것이 스트림이라는 겁니다.

너무 많은 컨셉들 (5:54)

반응형 프로그래밍은 어렵지만 계속 위세가 높아지고 있습니다. React Native나 React 등 여러 라이브러리가 유행을 선도하고 있죠. Cycle이나 Elm처럼 새로운 툴도 있고 반응형 프로그래밍에 초점을 맞춘 다른 언어도 있습니다.

그러나 이해해야할 컨셉이 너무 많은 것은 사실입니다. 제가 가장 중요하다고 생각하는 것은 observables와 observers입니다. subscriber, subscription, producer, hot/cold observables, backpressure, scheduler, subject 등 다른 컨셉도 많은데 제가 언급한 것은 전체의 10%도 안될 겁니다. 할 일이 태산처럼 느껴지죠?

너무 많은 컨셉에 혼란스럽더라도 괜찮습니다. 겁내지 마시고 계속해서 노력하면 언젠가는 이해할 날이 반드시 옵니다.

적용 과정에서의 애로 사항 (7:15)

이제 Airbnb에서의 적용 과정이나 배운점, 잘된 점과 아쉬운 점이 무엇인지 말씀드리겠습니다.

팀 규모 - 저희 팀은 15명이라는 꽤 큰 규모로, 여태까지 제 경험 중 가장 팀원이 가장 많았습니다. 모든 사람들이 안드로이드 앱의 코드를 생산해서 같은 저장소에 올렸죠. 하나의 코드를 기반으로 모든 팀원이 다른 팀원의 코드를 리뷰했습니다. 모든 사람이 무슨 일이 일어나는지 파악할 수 있도록 작성된 모든 코드를 이해하는 것이 중요했습니다.

코드 리뷰에는 Phabricator를 사용했는데 GitHub의 pull request 과정과 유사합니다. 코멘트를 달거나 제안을 하고 피드백을 주는 등의 여러 리액션이 가능하죠. RxJava를 적용하기로 결심하기 이전에 모든 팀원이 같은 입장이 되도록 해야 합니다. 만약 이런 공감없이 그냥 작성을 시작한다면 나머지 사람들은 무슨 일이 벌어지는지 이해할 수 없으므로 보이는 코드 내부에서 어떤 작용이 일어나는지 알아채기 어렵습니다. 팀원이 달랑 두 명이라면 괜찮겠지만 그 이상이라면 모든 사람들이 함께 따라갈 수 있도록 이해의 수준을 맞추기가 정말 어렵죠.

학습 곡선 - 이해해야 할 것이 많으므로 어이없는 실수를 저지를 때도 많습니다. 말도 안되는 코드를 작성하거나 상용 코드에서 크래쉬를 일으키기도 하겠지만 결국 괜찮아질 겁니다. 제 경험상 모든 사람들이 이해하는데는 약 두 달 가량이 걸리더군요. RxJava를 도입하고 싶다면 일단 그룹으로 이에 대해 논의하고 나머지 팀원에게 이들 컨셉을 설명하는 것을 추천합니다. 스스로의 이해도를 높이고 어느 정도 자신이 붙은 다음 다른 사람들을 모두 모아 직접 안드로이드 스튜디오를 열어서 예제 코드를 보여주면서 설명해보세요.

디버깅 - 정말 큰 문제죠. RxJava 커뮤니티의 모두들 디버깅의 중요성과 코드 수정의 필요성을 알고 있습니다. 최근 저희 버그 트래킹 시스템에서 스택 트레이스를 했는데 정말 많은 예외들이 있더군요. 많은 노이즈가 있는 엄청난 작업량이었습니다. 실제로 이와 관련된 작업을 하는 분이 있는지는 모르겠지만 빨리 고쳐졌으면 좋겠습니다.

흔히 저지르는 실수들 (11:34)

이제부터는 저희가 RxJava를 적용하는데 방해 요소가 된 것들을 공유하고자 합니다.

observeOn() (11:49)

RxJava를 사용하려면 이 핵심 컨셉을 이해해야 합니다.

return observableFactory.<T>toObservable(this)
	 .compose(this.<T>transform(observableRequest))
	 .observeOn(Schedulers.io())
	 .map(new ResponseMetadataOperator<>(this))
	 .flatMap(this::mapResponse)
	 .observeOn(AndroidSchedulers.mainThread())
	 .<AirResponse<T>>compose(group.transform(tag))
	 .doOnError(new ErrorLoggingAction(request))
	 .doOnError(NetworkUtil::checkForExpiredToken)
	 .subscribe(request.observer());

위 예제는 저희 앱에서 RxJava의 observable 스트림을 만드는데 사용하는 코드입니다. observeOn를 두 번 부르는 것이 낯설 겁니다. observeOn를 호출할 때마다 이 시점 이후의 모든 것이 해당 스케쥴러에서 실행되고 다음 번 호출에서 다시 전환됩니다.

RxJava를 사용하면 스트림을 만들게 됩니다. 흔히 RxJava가 비동기식이라고 오해하기 쉬운데, 사실상 기본 설정은 동기적입니다. 스트림을 만드는 것은 어느 시점에 실제로 subscribe할 것인지 결정하는 것뿐이고, subscribe를 하는 시점에 이를 함께 묶어서 실제로 실행하게 됩니다. 즉, subscribe를 호출하기 전에는 단지 스트림을 만들 뿐이죠. 선언 과정과 비슷합니다. observeOn를 부르면 다른 스레드로 전환합니다. 만약 observeOn를 부르지 않으면 observable을 subscribe한 동일 스레드에서 실행되겠죠. 따라서 observeOn은 작업의 로드를 다른 스레드로 나눌 수 있는 효과적인 방법이자 비동기적으로 작용하는 부분입니다.

observeOn을 처음 호출하면 스케쥴러를 넘깁니다. I/O 스케쥴러를 포함한 몇몇 스케쥴러가 이미 포함돼 있죠. 당연히 I/O 스케쥴러는 I/O에 묶인 스레드 풀인 I/O 스레드에서 작용합니다. map과 flatMap 오퍼레이터가 이 스레드를 실행하며, 작업이 끝나면 이를 다시 메인 스레드로 보냅니다. 따라서 메인 스레드에서 작업하면서 이들이 메인 스레드에서 호출한 것처럼 가정하면서 백그라운드로 보내고, 작업이 끝나면 다시 메인 스레드로 보낼 수 있습니다.

RxJava를 사용하지 않는다면 상당히 복잡한 작업이지만, 뭘 할 것인지 짧고도 단순하고 선언하는 방식만으로도 이를 수행할 수 있습니다. 바로 이 때문에 RxJava가 복잡해보이기도 합니다. 코드 양은 짧지만 내부적으로 무슨 일이 일어나는지 파악하는 것에는 많은 시간이 걸리니까요.

subscribeOn() (16:14)

subscribeOn은 observeOn과 밀접한 관계에 있는 컨셉입니다. observable이 subscribe한 스레드를 변경하는데, 용어에 익숙하지 않다면 좀 복잡하게 들릴 겁니다.

return observableFactory.<T>toObservable(this)
	 .compose(this.<T>transform(observableRequest))
	 .observeOn(Schedulers.io())
	 .map(new ResponseMetadataOperator<>(this))
	 .flatMap(this::mapResponse)
	 .observeOn(AndroidSchedulers.mainThread())
	 .<AirResponse<T>>compose(group.transform(tag))
	 .doOnError(new ErrorLoggingAction(request))
	 .doOnError(NetworkUtil::checkForExpiredToken)
	 .subscribeOn(Schedulers.io())
	 .subscribe(request.observer());

처음 호출인 observableFactory.<T>toObservable은 observable 객체가 만들어지는 곳으로 실제로 subscribeOn로부터 영향을 받습니다. subscription과 관련된 코드는 여러 종류가 있는데, subscription 상에서 실행되는 코드와 subscribe를 하면서 실행되는 코드가 있고, 해당 스트림에서 변화된 다른 코드들도 있죠. 이 중 subscribeOn은 어디서 호출하건 상관없이 subscription이 실행되는 스레드만 바꿉니다.

에러 핸들링 (18:05)

return observableFactory.<T>toObservable(this)
	 .compose(this.<T>transform(observableRequest))
	 .observeOn(Schedulers.io())
	 .map(new ResponseMetadataOperator<>(this))
	 .flatMap(this::mapResponse)
	 .observeOn(AndroidSchedulers.mainThread())
	 .<AirResponse<T>>compose(group.transform(tag))
	 .doOnError(new ErrorLoggingAction(request))
	 .doOnError(NetworkUtil::checkForExpiredToken)
	 .subscribeOn(Schedulers.io())
	 .subscribe(request.observer());

에러 로그를 위해 doOnError를 사용했습니다. 네트워크 통신이 실패할 때 분석 서비스에 로그를 남기고 몇 번이나 실패가 발생하는지 알고 싶다고 가정해 봅시다. doOnError는 스트림에서 에러를 받을 때마다 실행되는 액션으로 여러 번 발생할 수 있으므로 하나의 스트림에 여러 번의 에러 핸들링을 할 수 있습니다. doOnError가 에러 이벤트를 발견하면 설정된 액션을 호출하는데, 이는 단지 사이드 이펙트에 불과합니다.

return observableRequest
	 .rawRequest()
	 .<Observable<Response<T>>>newCall()
	 .observeOn(Schedulers.io())
	 .unsubscribeOn(Schedulers.io())
	 .flatMap(responseMapper(airRequest))
	 .onErrorResumeNext(errorMapper(airRequest));

catch 블럭처럼 작용하는 onErrorResumeNext를 사용할 수도 있는데 사실 반응형 프로그래밍과는 어울리지 않습니다. 마치 에러를 발견할 때마다 ‘이 에러를 catch하고 예외를 래핑해서 로그하고 빈 데이터를 리턴하는 액션을 여기서 실행해야지.’’ 하는 것과 마찬가지입니다. 명령형 프로그래밍처럼 생각한다면 catch 블럭처럼 작용할 수밖에 없죠.

유닛 테스트 (20:04)

비동기 데이터 스트림에서 유닛 테스트를 한다는 것은 복잡한 일이므로 RxJava는 TestSubscriber라는 클래스를 지원합니다.

@Test public void testErrorResponseNonJSON() {
 server.enqueue(new MockResponse()
	 .setBody("something bad happened")
	 .setResponseCode(500));
 TestRequest request = new TestRequest.Builder<String>().build();
 TestSubscriber<AirResponse<String>> subscriber = new TestSubscriber<>();
 observableFactory.<String>toObservable(request).subscribe(subscriber);
 subscriber.awaitTerminalEvent(3L, TimeUnit.SECONDS);
 NetworkException exception = (NetworkException)
	 subscriber.getOnErrorEvents().get(0);
 assertThat(exception.errorResponse(), equalTo(null));
 assertThat(exception.bodyString(), equalTo("something bad happened"));
}

TestSubscriber를 사용해서 스트림을 subscribe하고 스레드를 블럭해서 이벤트를 받을 때까지 기다릴 수 있습니다. 예를 들어 .awaitTerminalEvent 메서드는 onCompleted나 onError 등 끝내는 이벤트가 올 때까지 스레드를 블럭합니다. 스트림의 모든 이벤트마다 onNext를 여러번 만나게 되며, 이벤트가 끝나면 성공이나 실패에 따라 onCompleted나 onError를 받습니다. 마침내 더 이상 어떤 이벤트도 받을 수 없게 되면 스트림을 끝내게 되죠.

@Test public void testUnicodeHeader() {
 server.enqueue(new MockResponse().setBody("\"Hello World\""));
 TestRequest request = new TestRequest.Builder<String>()
 .header("Bogus", "中華電信")
 .build();
 observableFactory.toObservable(request)
 .toBlocking()
 .first();
 RecordedRequest recordedRequest = server.takeRequest();
 assertThat(recordedRequest.getHeader("Bogus"), equalTo("????"));
}

스레드를 블럭하는 다른 방법으로 toBlocking을 사용할 수도 있는데, 유닛 테스트에는 좋지만 상용 코드에는 그다지 유용하지 않습니다. RxJava를 사용한다면 스레드를 블럭하는 것을 좋아하지 않겠지만 테스트에는 유용하죠. test subscriber 사용 코드보다 짧고, 언제 실패하지 않는지 알 수 있다면 블럭해서 첫번째 이벤트만을 받을 수 있습니다.

저희는 OkHttp를 사용해서 가상 응답을 만들어서 가짜 응답을 보냈습니다. OkHttp에 버그가 있으면 어떻게 될지 특별한 캐릭터를 헤더를 붙여 테스트를 했습니다. 헤더를 잘 지우는지 테스트한 거죠.

메모리 릭 (22:41)

모바일 프로그래밍에서는 메모리 예외를 아는 것이 중요합니다.

스트림을 subscribe하면 subscription을 되돌려 받습니다. subscription을 받으면 이를 unsubscribe해서 자원을 회수할 수 있죠. 이 경우 해당 스트림에 대한 레퍼런스는 더 이상 가지지 않습니다. 안드로이드 액티비티나 프래그먼트 등에서 요청을 만들 때 중요한 일입니다. 단지 버리고 잊는 것으로는 부족하며, 액티비티가 없어질 때 할당된 자원을 비우는 것이 좋습니다. 꽤 자주 만들게 되는 패턴이죠.

private final CompositeSubscription pendingSubscriptions =
	new CompositeSubscription();

@Override public void onCreate() {
  pendingSubscriptions.add(
	  observable.subscribe(observer));
}

@Override public void onDestroy() {
  pendingSubscriptions.clear();
}

여러 subscription을 함께 그룹으로 묶는 CompositeSubscription 클래스를 사용하면 편리합니다. 추가적인 subscription이 생기면 이 그룹에 추가할 수도 있죠. 액티비티를 없앤 후 이 클래스 역시 비우면 됩니다.

참고 자료 (23:55)

• Reactive.community
• Reactive Extensions at Netflix
• Ben Christensen’s talk on the same subject