새로운 자바 SDK를 위해 카우치베이스가 RxJava를 선택한 이유

이 블로그 게시물은 새로운 Java SDK의 핵심 구성 요소 중 하나로 RxJava를 선택하게 된 배경과 동기를 설명합니다.

동기부여

API를 설계하는 방법은 매우 다양하며, 저마다 고유한 장점(과 단점)을 가지고 있습니다. 완전히 새로운 API를 설계하는 과정에서 주요 질문 중 하나는 이를 사용자에게 어떻게 노출할 것인가였습니다. 

우리가 스스로에게 물어볼 필요가 없었던 한 가지 질문은 바로 "동기식으로 해야 할까, 아니면 비동기식으로 해야 할까?"였습니다. 우리는 비동기식 API가 매우 자주 필요로 하는 성능과 확장성을 얻을 수 있는 유일하고도 타당한 방법이라고 굳게 믿으며, 또한 비동기에서 동기로 전환하는 것이 그 반대보다 훨씬 쉽습니다. 현재 안정 버전의 SDK(이 글을 작성하는 시점 기준 1.4.3)는 비동기 응답을 제공하기 위해 이미 다양한 방식으로 Future를 적극적으로 활용하고 있으며, 이는 spymemcached가 API에 해당 개념을 처음 도입했던 2006/2007년까지 거슬러 올라갑니다.

자바의 Future 인터페이스가 (스칼라의 future 같은) 다른 솔루션들에 비해 매우 제한적이라는 것은 잘 알려져 있습니다. 게다가 하나의 연산이 다른 연사에 의존하고 전체 과정을 비동기로 유지하고 싶은 비동기 데이터 플로우를 구축해야 하는 경우 코딩하기가 다소 더 까다로워집니다. 최근 버전에서는 리스너에 대한 지원을 추가하여 상황이 꽤 나아졌지만 여전히 이상적인 솔루션은 아닙니다.

지난 몇 년 동안, 우리는 밀접하게 주시해 온 다른 라이브러리들과 패턴들이 등장했습니다. 성숙한 개념 중 하나는 마이크로소프트와 .NET에서 유래한 Reactive Extensions로 알려져 있습니다. 이는 애플리케이션이 이벤트 지향적이어야 하며 비동기 방식으로 그러한 이벤트에 반응해야 한다는 아이디어를 기반으로 합니다. 이 개념은 데이터로 할 수 있는 일(수정, 결합, 필터링 등)에 대한 매우 풍부한 연산자 집합을 정의합니다. 최근 넷플릭스는 이를 자바로 포팅하여 RxJava라는 별칭을 붙였습니다(현재 이 프로젝트는 넷플릭스 네임스페이스 하에 있지만, 조만간 “io.reactivex”로 이동될 예정입니다). 이 라이브러리는 매우 안정적이며, 지원 범위를 넓히려는 우리의 계획과도 잘 부합하는 스칼라, 그루비, JRuby와 같은 다른 JVM 언어용 어댑터도 제공합니다.

개념

Rx의 핵심 아이디어는 Observable과 그 관찰자(observer)를 중심으로 돌아갑니다. 이 개념을 접해보지 않았다면, Observable을 Iterable의 비동기 및 푸시 기반 사촌(또는 더 공식적으로는 쌍대(dual))으로 생각할 수 있습니다. 더 구체적으로, 이들의 관계는 다음과 같습니다:

이벤트	이터러블(풀)	옵서버블 (푸시)
데이터 검색	T 다음에()	onNext(T)
오류 발견	예외를 던진다	onError(예외)
완료	반품	완료됨()

데이터가 Observable로 푸시될 때마다, 구독 중인 모든 observer는 onNext() 메서드에서 해당 데이터를 받습니다. Observable이 결국 완료되면 (항상 그런 것은 아니지만) onCompleted 메서드가 호출됩니다. 이제 프로세스 중 어디에서든 에러가 발생하면 onError 메서드가 호출되며, 해당 Observable은 완료된 것으로 간주됩니다.

문법을 좋아하신다면, 계약서는 이렇게 생겼습니다: 

OnNext* (OnCompleted | OnError)?

특히 1개 또는 N개의 데이터가 반환되는지에 대한 구분이 없다는 점을 명시해 두세요. 이는 호출하는 메서드와 해당 문서에 기술된 내용을 통해 자연스럽게 유추할 수 있습니다. 어차피 프로그래밍 흐름에는 변화가 없습니다. 조금 추상적이므로 구체적인 예시를 살펴보겠습니다. CouchbaseCluster 클래스에는 필요한 모든 리소스를 초기화한 다음 작업할 수 있는 Bucket 인스턴스를 반환하는 openBucket이라는 메서드가 있습니다. 이제 소켓을 열고, 설정을 가져오는 등의 작업에는 다소 시간이 걸릴 수 있으므로, 이는 완벽한 후보가 됩니다. 블로킹 API는 다음과 같은 형태를 띨 것입니다:

인터페이스 Cluster {
        Bucket openBucket(String name, String password);
}

어떻게 비동기로 만들 수 있을까요? Observable로 감싸야 합니다:

인터페이스 Cluster {
        관찰 가능성

openBucket(String name, String password);
}

따라서 이제 우리는 사용할 수 있는 버킷 인스턴스를 결국 반환하는 옵저버블을 반환합니다. 옵저버를 추가해 봅시다:

클러스터.버킷 열기().구독(새로운 관찰자양동이<() {
    @Override
    공공의 무효 완료됨() {
        시스템.밖으로.println(“옵저버블 완료!”);
    }

    @Override
    공공의 무효 오류 발생 시(던질 수 있는 것 e) {
        시스템.오류.println(“무언가 일어났다”);
        e.스택 추적 출력();
    }

    @Override
    공공의 무효 다음(양동이 양동이) {
        시스템.밖으로.println(“버킷 수신함: “ + 양동이);
    }
});

이러한 메서드들은 다른 스레드에서 호출되므로, 코드를 이 상태로 두고 메인 스레드를 바로 종료하면 아무것도 볼 수 없을 것입니다. 이제 나머지 모든 코드를 onNext 메서드 안에 작성할 수도 있지만, 그것이 아마 가장 좋은 방법은 아닐 것입니다. 버킷은 대개 
처음부터 공개하고 싶었는데, 그것을 블로킹한 뒤 나머지 코드를 진행할 수 있습니다. 모든 Observable은 Iterable처럼 느껴지는 블로킹 Observable로 변환할 수 있습니다.

블로킹오블저버블

blockingObservable = cluster.openBucket().toBlocking();

수신된 데이터에 대해 블로킹 방식으로 반복 작업을 수행하는 여러 가지 방법을 찾을 수 있지만, 단 하나의 값만 예상되는 경우(우리 경우가 바로 그렇습니다) 사용할 수 있는 단축 메서드도 있습니다.

Bucket bucket = cluster.openBucket().toBlocking().single();

여기서 내부적으로 일어나는 일은 onNext에서 호출된 값이 우리를 위해 저장되고 onComplete가 호출되면 한 번 반환된다는 것입니다. onError가 호출되면 throwable이 직접 던져지므로 이를 잡을 수 있습니다.

API 통합

이제 여러분이 보신 것은 새발의 피에 불과합니다. 버킷 개방 역시 퓨처(Future)로 처리할 수도 있습니다. 혼자서입니다. Observables가 진가를 발휘하는 때는 둘 이상의 결과값이 반환되는 경우를 다루어야 할 때이며, 이 경우 Future는 더 이상 목적에 부합하지 않으며 미래또는 이와 유사한 것은 동일한 계약을 갖지 않습니다. Observable은 둘 이상의 T가 반환될 수 있음을 암시하므로, 때로는 하나의 T가 반환되고 때로는 둘 이상의 T가 반환되더라도 API는 동일해 보일 수 있습니다.

구체적인 예를 다시 살펴보겠습니다. SDK는 하나의 문서를 반환하는 get 메서드를 제공합니다. 다음과 같습니다:

인터페이스 Bucket {
        관찰 가능성

get(String id);
}

하지만 저희는 또한 두 개 이상의 결과(또는 결과가 전혀 없을 수도 있음)를 잠재적으로 반환하는 쿼리(뷰, N1QL)도 지원합니다. Observable 계약 덕분에 다음과 같은 API를 구축할 수 있습니다:

인터페이스 Bucket {
        관찰 가능성

query(ViewQuery query);
}

보세요? Observable이 어떻게 동작해야 하는지 알고 있기 때문에, 계약에는 암묵적으로 “쿼리를 전달하면 N개의 ViewResults를 돌려받는다”고 적혀 있는 셈입니다. 그리고 더 큰 그림에서 보면, 여러분이 기대하는 방식으로 직관적으로 동작하는 메서드들이 훨씬 더 많습니다.

인터페이스 양동이 {
    D 확장합니다 문서>> 관찰 가능성D< 삽입하다(D 문서);
    D 확장합니다 문서>> 관찰 가능성D< 업서트(D 문서);
    D 확장합니다 문서>> 관찰 가능성D< 교체하다(D 문서);

    관찰 가능성뷰 결과< 질의(ViewQuery 쿼리);
    관찰 가능성검색 결과< 질의(쿼리 쿼리);
    관찰 가능성검색 결과< 질의(문자열 질의);

    관찰 가능성부울< 비우다();
}

데이터플로우를 비동기화해 주세요!

지금까지 우리는 옵저버블이 무엇을 할 수 있는지, 그리고 결합력 있고 단순하면서도 비동기적인 API를 제공하는 데 어떻게 도움을 주는지 살펴보았습니다. 그러나 옵저버블은 진정으로 그들의 조합 가능성(composability) 측면에서 빛을 발합니다. 옵저버블로는 정말 많은 것들을 할 수 있으며, 이 게시물에서 그 전부를 다 다룰 수는 없습니다. RxJava에는 여기에 있는 아주 훌륭한 참고 문서가 있으니 꼭 확인해 보세요. 이 문서는 비동기 데이터 흐름이 어떻게 작동하는지 보여주기 위해 마블 다이어그램(marble diagram)을 사용하고 있으며, 이는 우리도 향후 문서의 일부로 제공하고자 하는 것입니다.

실용적인 예를 들어보겠습니다. 카우치베이스(사용자 세부 정보가 포함된 완전한 JSON 객체)에서 문서를 로드하려고 하지만, 코드 뒷부분에서는 이름(firstname)만 가지고 어떤 작업을 수행하고 싶은 상황입니다. 이때 map 함수를 사용하여 JsonDocument를 firstname 문자열(String)로 매핑할 수 있습니다.

양동이
    .얻다(“user::1”)
    .지도(새로운 함수1JsonDocument, 문자열<() {
        @Override
        공공의 문자열 전화(JsonDocument jsonDocument) {
            반환 JSON 문서.콘텐츠().문자열가져오기(“이름”);
        }
    })
    .구독(새로운 동작 1문자열<() {
        @Override
        공공의 무효 전화(문자열 이름) {
            시스템.밖으로.println(이름);
        }
    });

여기에는 두 가지 중요한 측면이 있습니다: 여기에 체인으로 연결된 모든 메서드도 비동기적으로 실행되므로 원래 스레드를 차단하지 않습니다. 카우치베이스(Couchbase)에 대한 get 호출이 반환되면 JSON 문서에서 firstname을 매핑한 다음 마지막으로 출력합니다. 완전한 Observer를 제공할 필요는 없습니다. onNext 값에만 관심이 있는 경우 (여기서 보여준 것처럼) 해당 값만 구현하면 됩니다. 더 많은 예제는 오버로드된 메서드를 참조하십시오.

또한 여기서는 의도적으로 Java 6/7 스타일의 익명 클래스를 보여주고 있다는 점을 참고하세요. 저희는 Java 8도 지원하지만, 그에 대해서는 나중에 더 다루겠습니다. 자, 만약 이름이 “a”로 시작하는 경우에만 이름을 출력하고 싶다면 이 체인을 어떻게 확장할 수 있을까요?

양동이
    .얻다(“user::1”)
    .지도(새로운 함수1JsonDocument, 문자열<() {
        @Override
        공공의 문자열 전화(JsonDocument jsonDocument) {
            반환 JSON 문서.콘텐츠().문자열가져오기(“이름”);
        }
    })
    .필터(새로운 함수1문자열, 부울<() {
        @Override
        공공의 부울 전화(문자열 s) {
            반환 s.시작하는(“a”);
        }
    })
    .구독(새로운 동작 1문자열<() {
        @Override
        공공의 무효 전화(문자열 이름) {
            시스템.밖으로.println(이름);
        }
    });

물론 단순한 if문으로도 충분하겠지만, 필터링을 위한 코드가 훨씬 더 복잡해질 수 있으며 (아마도 다른 뭔가를 호출할 수도 있음) 상상할 수 있습니다. 옵저버블 변환에 대한 마지막 예제로, 우리는 매우 자주 발생하는 작업을 수행할 것입니다: 문서를 로드하고, 그 내용을 수정한 다음, 다시 카우치베이스(Couchbase)에 저장하는 것입니다:

양동이
    .얻다(“user::1”)
    .지도(새로운 함수1JsonDocument, JsonDocument<() {
        @Override
        공공의 JsonDocument 호출(JsonDocument 원본) {
            원본.콘텐츠().넣다(“이름”, “SomethingElse”);
            반환 오리지널;
        }
    })
    .플랫맵(새로운 함수1JsonDocument, ObservableJsonDocument>>() {
        @Override
        공공의 관찰 가능성JsonDocument< 전화(JsonDocument가 수정되었습니다) {
            반환 양동이.교체하다(수정됨);
        }
    }).구독();

FlatMap은 map과 매우 유사하게 동작하며, 차이점은 자체적으로 Observable을 반환하므로 비동기 작업을 매핑하는 데 완벽하게 적합하다는 점입니다.

또 다른 측면은 Observable을 사용하면 정교한 에러 처리를 손쉽게 할 수 있다는 점입니다. 2초의 타임아웃을 적용하고 호출이 반환되지 않으면 대신 다른 것을 반환하는 예제를 구현해 보겠습니다.

양동이
    .얻다(“user::1”)
    .시간 초과(2, TimeUnit.초)
    .onErrorReturn(새로운 함수1던질 수 있는 것, JsonDocument<() {
        @Override
        공공의 JsonDocument 호출(던질 수 있는 것 던질 수 있는) {
            반환 JsonDocument.만들다(“user::anonymous”, JsonObject.비어 있음().넣다(“이름”, “존 도”));
        }
    });

여기서 get 호출이 2초 안에 반환되지 않으면 (예제를 위해 합리적인 기본값을 가정하는) 더미 문서가 반환됩니다. 이것은 단순한 예시일 뿐이지만, 재시도, 다른 옵저버블(Observable)로의 분기 등 예외를 활용해 많은 것을 할 수 있습니다. 올바른 사용법은 공식 문서(및 Rx 문서)를참고하시기 바랍니다.

잠깐, 더 있습니다

서로 다른 옵저버블의 결합(병합, 지핑, 연결), 시간 간격에 따른 결과 일괄 처리, 부수 효과 수행 등 훨씬 더 많은 기능을 사용할 수 있습니다. 개념을 이해하는 초기(작은) 난관을 넘어서고 나면 매우 자연스럽게 느껴지며, 다시는 돌아가고 싶지 않을 것이라고 약속드립니다(만약 저희가 틀렸다면, 언제든지 옵저버블을 블로킹하거나 퓨처로 변환할 수 있습니다).

RxJava는 훌륭한 Java 8 지원 기능도 갖추고 있으므로, 이미 프로젝트에서 이를 사용할 수 있는 행운아라면 위의 예제를 다음과 같이 단순화할 수 있습니다.

양동이
    .get(“user::1”)
    .map(jsonDocument -> jsonDocument.content().getString(“firstname”))
    .filter(s -> s.startsWith(“a”))
    .subscribe(System.out::println);

멋지죠? RxJava는 그 위에 다양한 언어 어댑터도 제공하는데, 이 글을 작성하는 시점 기준으로 Scala, Clojure, Groovy, JRuby, Kotlin이 있습니다. 이 어댑터들을 사용하면 더욱 언어에 특화된 통합을 제공할 수 있으며, 수요가 생기는 대로 각 언어별 Couchbase 지원을 강화하기 위해 이 중 일부를 활용할 계획입니다. 자바 SDK를 제외하고 우리의 최우선 순위는 단연코 스칼라(Scala)이므로, 조만간 있을 관련 발표들을 기대해 주세요!

이제 저희만큼 기대가 크시기를 바라며, 평소 소통 채널을 통해 피드백과 질문을 보내주시기를 기대합니다!