ReactiveX Introduction 번역

ReactiveX란?

MS 사에서 만든 observable streams 기반의 비동기 프로그래밍 API

reactivex.io 에서는 ReactiveX를 다음과 같이 설명합니다.

ReactiveX는 옵저버 패턴 , 이터레이터 패턴 그리고 함수형 프로그래밍 의 장점(best ideas)들을 결합한 것입니다.

It extends the observer pattern to support sequences of data and/or events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.

ReactiveX는 옵저버 패턴을 확장해 데이터나 이벤트의 연속적인 흐름을 다룰 수 있도록 하고, 여러 시퀀스를 선언적으로 조합할 수 있는 연산자를 제공합니다. 이를 통해 개발자는 스레드 처리, 동기화, 스레드 안전성, 동시성 자료구조, 논블로킹 I/O와 같은 저수준 구현 문제를 직접 다루지 않고도 복잡한 비동기 로직을 표현할 수 있습니다.

Observable은 다양한 아이템들의 비동기 흐름에 접근하는 가장 이상적인 방식으로 그 격차를 메웁니다.

	Single items	multiple items
synchronous	T getData()	Iterable<T> getData()
asynchronous	Future<T> getData()	Observable<T> getData()

이것은 때때로 “함수형 반응형 프로그래밍(functional reactive programming)”이라고 불리기도 하지만, 이는 잘못된 명칭(misnomer)입니다. ReactiveX는 함수형일 수도 있고, 반응형일 수도 있지만, “함수형 반응형 프로그래밍”은 전혀 다른 개념입니다.
주요 차이점 중 하나는, 함수형 반응형 프로그래밍은 시간이 흐름에 따라 연속적으로 변하는 값들을 다루는 반면, ReactiveX는 시간이 지남에 따라 불연속적으로 발생(discrete)하는 값들을 다룬다는 것입니다.

왜 Observable을 사용하나요?

Rx의 Observable 모델은 배열과 같은 데이터 컬렉션을 다룰 때 사용하는 간단하고 조합 가능한 연산자들과 동일한 방식으로 비동기 이벤트 스트림을 다룰 수 있게 해줍니다. 또한 복잡하게 얽힌 콜백 구조에서 벗어나도록 도와주며, 그 결과 코드가 더 읽기 쉬워지고 버그가 발생할 가능성도 줄어듭니다.

Observable은 조합 가능(Composable)합니다.

Java의 Futures 같은 단일 비동기 실행에는 간단히 쓸 수 있지만, 여러 단계로 중첩되면 금방 복잡성이 커집니다.
Futures를 이용해 조건부 비동기 실행 흐름을 최적으로 구성하기는 어렵습니다.
심지어 각 요청의 지연 시간(latency)이 실행 시점마다 달라지므로 사실상 불가능 할 수도 있습니다.
(→ 즉, “만약 이 결과가 true인 경우 다음 요청 실행” 같은 조건부 플로우를 짜기가 힘들고, 요청 지연시간이 불규칙해서 더 어려워짐)
물론 할 수는 있지만, 그렇게 하면 금방 복잡해져서 에러가 생기기 쉽거나, 아니면 Future.get() 을 너무 일찍 호출해서 차라리 동기 실행처럼 되버립니다.
그 결과 비동기의 장점(논블로킹)이 사라져 버립니다.
반대로 ReactiveX의 Observables는 비동기 데이터 흐름과 시퀀스를 조합하기 위해 설계된 것입니다.
(즉, Futures 처럼 단일 작업을 다루는 게 아니라, 여러 비동기 흐름을 조합/연결하는 데 최적화되어 있습니다.

Observable은 유연(Flexible)합니다.

ReactiveX의 Observable은 (Future 가 하는 것처럼) 단일 값하나의 방출만을 지원하는 것이 아니라, 값들의 연속된 시퀀스나 심지어 무한 스트림도 지원합니다. Observable 은 이 모든 경우에 사용할 수 있는 단일한 추상화이며, 이는 Iterable이 가진 유연함과 우아함을 모두 갖추고 있습니다.

Observable은 특정 방식에 얽매이지 않습니다.

ReactiveX는 특정한 동기/비동기 처리 방식에 치우치지 않습니다. Observable은 스레드 풀, 이벤트 루프, 논블로킹 I/O, 엑터 모델(Akka 등) 등, 어떤 방식으로든 구현될 수 있습니다. 즉, 개발자의 필요, 스타일, 전문성에 맞는 방식이면 무엇이든 사용할 수 있습니다. 클라이언트 코드 입장에서는, Observable과 상호작용할 때 항상 비동기적으로 보인다는 것입니다. 실제 내부 구현이 blocking인지 non-blocking인지 무관하게 말입니다.

Observable은 Iterable의 쌍둥이 같은 개념으로, Iterable이 동기적·pull 기반이라면 Observable은 비동기적·push 기반입니다.

event	Iterable (pull)	Observable (push)
retrieve data	T next()	onNext(T)
discover error	throws Exception	onError (Exception)
complete	!hasNext()	onCompleted()

Observable은 어떻게 구현되어 있나요?

public Observable<data> getData();

• 호출자와 같은 스레드에서 동작하나요?
• 다른 스레드에서 비동기적으로 동작하나요?
• 여러 개의 스레드로 일을 나눠서, 순서에 상관없이 데이터를 돌려주나요?
• 스레드 풀 대신 엑터를 사용하나요?
• NIO + 이벤트 루프로 네트워크를 비동기 처리하나요?
• 이벤트 루프로 작업 스레드와 콜백 스레드를 분리하나요?

관찰자 입장에서는 중요하지 않습니다!

그리고 중요한 점: ReactiveX에서는 나중에 마음이 바뀌어도, Observable의 내부 구현 방식을 과감히 바꾸면서도, 그 Observable을 소비하는 코드(클라이언트)를 깨뜨리지 않을 수 있습니다.

콜백에도 나름의 문제가 있습니다.

콜백은 어떤 것도 블로킹하지 않음으로써 Future.get()에서 발생하는 조기 블로킹 문제를 해결합니다. 콜백은 응답이 준비되었을 때만 실행되므로, 본질적으로 효율적입니다. 하지만 Future와 마찬가지로, 콜백도 단일 레벨의 비동기 실행에는 사용하기 쉽지만, 중첩된 구조로 합성되면 다루기 어렵고 복잡해집니다.

ReactiveX는 다언어(Polyglot) 구현체입니다.

ReactiveX는 여러 프로그래밍 언어에서 구현된 라이브러리로, 각 언어의 문법적 특성과 관용적 스타일을 반영합니다. 또한 현재도 빠른 속도로 더 많은 언어로 확장되고 있습니다.

Reactive Programming

ReactiveX는 Observable을 다룰 수 있는 다양한 연산자(filter, select, transform, combine 등)를 제공합니다. 이를 통해 효율적인 실행과 조합이 가능합니다. Observable 클래스는 Iterable의 “pull” 방식에 대응하는 “push” 방식이라고 생각하면 됩니다. Iterable에서는 소비자(consumer)가 생산자로부터 값을 끌어오며(pull), 값이 준비될 때까지 스레드가 블로킹 됩니다. 반대로, Observable에서는 생산자(producer)가 값이 준비되는 즉시 소비자에게 값을 밀어(push)줍니다. 이 방식은 값이 동기적으로든 비동기적으로든 도착할 수 있기 때문에 더 유연합니다.

Iterable과 Observable 모두에 map, filter 같은 고차함수를 적용할 수 있는데, 그 유사성을 보여주는 예제 코드

Iterable

getDataFromLocalMemory()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .forEach({ println "next => " + it })

Observable

getDataFromNetwork()
  .skip(10)
  .take(5)
  .map({ s -> return s + " transformed" })
  .subscribe({ println "onNext => " + it })

Observable 타입은 기존 GoF Observable 패턴에 부족했던 두 가지 의미(sementics)를 추가하여, Iterable 타입에서 제공되는 것과 대응되도록 합니다.

1. 생산자가 소비자에게 더 이상 데이터가 없음을 알릴 수 있는 기능
   • Iterable 에서는 foreach 루프가 정상적으로 끝나는 것이 데이터 없음 신호
   • Observable 에서는 onComplete() 호출
2. 생산자가 소비자에게 에러 발생을 알릴 수있는 기능
   • Iterable 에서는 순회중 예외가 발생하면 throw
   • Observable 에서는 onError() 호출

이 두 가지가 추가됨으로써, ReactiveX는 Iterable과 Observable 타입을 조화시킵니다. 둘의 유일한 차이는 데이터가 흐르는 방향 입니다. 이것은 매우 중요한데, 이제 Iterable에서 할 수 있는 모든 연산을 Observable에서도 동일하게 할 수 있기 때문입니다.