<div id="readability-page-1" class="page"><section>
            <p>현재 약 2년동안 실무에 RxJava를 사용하고 있지만, 아직까지(부끄럽게도) 항상 동작이 헷갈리는 함수 2개가 있다. <br>
바로 <code>subscribeOn()</code> 과 <code>observeOn()</code>이다. 이미 많은 블로그들의 글을 읽어봤지만, 여전히 헷갈리는 듯하여 RxJava 내부적인 구현의 관점에서 나름대로 한번 정리를 해볼려고 한다.</p>

<hr>

<h3 id="reactivestreams">Reactive Streams</h3>

<p>먼저 RxJava의 <code>subscribeOn()</code> 과 <code>observeOn()</code>의 동작을 이해하기 위해서는, <strong><a id="jupiny-blog-link" href="https://github.com/reactive-streams/reactive-streams-jvm" target="_blank">Reactive Streams</a></strong> 의 기본 메커니즘을 이해하여야 한다. RxJava도 결국 이 Reactive Streams의 규격을 따르는 구현체의 하나일 뿐이기 때문이다.</p>

<h4 id="apispecification">API Specification</h4>

<p>Reactive Streams API 명세를 살펴보면, 아래와 같이 3개의 interface <code>Publisher</code>, <code>Subscription</code>, <code>Subscriber</code> 가 있다. (사실 <code>Processor</code>도 있지만, 여기서는 무시)</p>

<pre><code>public interface Publisher&lt;T&gt; {  
    // 새로운 Subscription을 하나 만들어 Subscribe의 onSubscribe() 호출 
    public void subscribe(Subscriber&lt;? super T&gt; s);
}

public interface Subscription {  
    // Publisher에게 n개의 데이터 발행 요청
    public void request(long n);
    // Publisher에게 데이터 발행 중단 요청
    public void cancel();
}

public interface Subscriber&lt;T&gt; {  
    // Subscription의 request() 호출
    // Publisher는 그에 대한 응답을 내려줌
    public void onSubscribe(Subscription s);
    // Subscription의 request() 요청에 대한 Publisher의 데이터 응답
    public void onNext(T t)
    // Publisher의 데이터 발행 중의 에러 공지
    public void onError(Throwable t);
    // Publisher의 데이터 발행 완료 공지
    public void onComplete();
}
</code></pre>



<h4 id="flow">Flow</h4>

<p><img src="https://d194zelh06zukz.cloudfront.net/2020/04/rxjava-subscribeon-observeon-1-1.png" alt=""></p>

<ol>
<li><code>Subscriber</code> 생성하여 <code>Publisher</code>의 <code>subscribe(Subscriber s)</code> 실행  </li>
<li><code>Publisher</code>의 <code>subscribe(Subscriber s)</code>: <code>Subscription</code> 생성하여 <code>Subscriber</code>의 <code>onSubscribe(Subscription s)</code> 실행  </li>
<li><code>Subscriber</code>의 <code>onSubscribe(Subscription s)</code>: <code>Subscription</code>의 <code>request()</code> 실행  </li>
<li><code>Subscription</code>의 <code>request()</code>: <code>Publisher</code>에게 데이터 요청  </li>
<li><code>Publisher</code>은 <code>Subscription</code>을 통해 <code>Subscriber</code>의 <code>onNext()</code>/<code>onError()</code>/<code>onComplete()</code> 실행</li>
</ol>

<hr>

<h3 id="rxjava2flowable">RxJava2의 Flowable 구현</h3>

<p>코드를 통해 직접 확인해보기 위해 Reactive Streams의 규격을 잘 따르고 있는 RxJava2의 <code>Flowable</code>의 구현을 한번 살펴보자. <br>
(<code>Single</code>, <code>Mabye</code> 의 경우는 Reactive Streams의 규격을 따르지 않고 있음.)</p>

<blockquote>
  <p>Rxjava2 2.2.19 버젼을 사용하였습니다.</p>
</blockquote>

<h4 id="subscribeon">subscribeOn</h4>

<pre><code>Flowable.just(1)  
        .subscribeOn(Schedulers.io()) // HERE
        .subcribe()
</code></pre>

<p><code>subscribeOn(Schedulers.io())</code> 코드의 구현을 따라가다 보면, <code>SubscribeOnSubscriber</code>이라는 <strong><a id="jupiny-blog-link" href="https://github.com/ReactiveX/RxJava/blob/v2.2.19/src/main/java/io/reactivex/internal/operators/flowable/FlowableSubscribeOn.java" target="_blank"><code>FlowableSubscribeOn</code></a></strong> 클래스의 내부 클래스를 만나게 된다. <br>
내부 코드 일부분을 살펴보면,</p>

<pre><code>@Override
public void onNext(T t) {  
    downstream.onNext(t); // NOTE
}

@Override
public void onError(Throwable t) {  
    downstream.onError(t); // NOTE
    worker.dispose();
}

@Override
public void onComplete() {  
    downstream.onComplete(); // NOTE
    worker.dispose();
}
</code></pre>

<p>일단 downstream의 <code>Subscriber</code>의 동작을 실행하는 부분에는 설정한 <code>Scheduler</code>로 별도로 스케줄링하지 않고 있다.</p>

<pre><code>@Override
public void request(final long n) {  
    if (SubscriptionHelper.validate(n)) {
        Subscription s = this.upstream.get();
        if (s != null) {
            requestUpstream(n, s); // NOTE
        } else {
            // ... 생략 ...
        }
    }
}

void requestUpstream(final long n, final Subscription s) {  
    if (nonScheduledRequests || Thread.currentThread() == get()) {
        s.request(n);
    } else {
        worker.schedule(new Request(s, n)); // NOTE
    }
}

static final class Request implements Runnable {  
    final Subscription upstream;
    final long n;

    Request(Subscription s, long n) {
        this.upstream = s;
        this.n = n;
    }

    @Override
    public void run() {
        upstream.request(n); // NOTE
    }
}
</code></pre>

<p><code>request()</code> 함수를 따라가보면, <code>scheduler.createWorker()</code>(위에 코드는 넣지 않았지만)로 생성한 <code>Worker</code>에 의해 upstream <code>Subscription</code>의 <code>request()</code>이 스케쥴링됨을 확인할 수 있다. </p>

<h4 id="observeon">observeOn</h4>

<pre><code>Flowable.just(1)  
        .observeOn(Schedulers.io()) // HERE
        .subcribe()
</code></pre>

<p>마찬가지로 <code>observeOn(Schedulers.io())</code> 코드의 구현을 따라가다 보면, <code>ObserveOnSubscriber</code>이라는 <strong><a id="jupiny-blog-link" href="https://github.com/ReactiveX/RxJava/blob/v2.2.19/src/main/java/io/reactivex/internal/operators/flowable/FlowableObserveOn.java" target="_blank"><code>FlowableObserveOn</code></a></strong> 클래스의 내부 클래스를 만나게 된다.  그리고 이 <code>ObserveOnSubscriber</code> 상속하고 있는 추상 클래스 <code>BaseObserveOnSubscriber</code> 를 살펴보면,</p>

<pre><code>@Override public final void onNext(T t) {
    // ... 생략 ...
    trySchedule(); // NOTE
}

@Override
public final void onError(Throwable t) {  
    // ... 생략 ...
    trySchedule(); // NOTE
}

@Override
public final void onComplete() {  
    // ... 생략 ...
    trySchedule(); // NOTE
}
</code></pre>

<p><code>Subscriber</code> 모든 함수의 마지막에 <code>trySchedule()</code>가 실행됨을 볼 수 있다. <code>trySchedule()</code> 의 구현을 보면,</p>

<pre><code>final void trySchedule() {  
    if (getAndIncrement() != 0) {
        return;
    }
    worker.schedule(this); // NOTE
}
</code></pre>

<p>역시 <code>scheduler.createWorker()</code>를 통해 생성된 <code>Worker</code>를 통해 실행되는데, 여기서 <code>this</code>는 <code>Runnable</code> 를 구현한 자신이다. <br>
그럼 <code>Runnable</code>를 구현한 부분을 보면,</p>

<pre><code>@Override
public final void run()  
    // ... 생략 ...
    runAsync(); // NOTE
}
</code></pre>

<p><code>runAsync()</code>이라는 추상 메쏘드를 실행한다. 처음으로 돌아와서 이 추상 클래스를 상속한 <code>ObserveOnSubscriber</code>에서 구현한 <code>runAsync()</code> 을 확인해보면,</p>

<pre><code>@Override
void runAsync() {  
    // ... 생략 ...
    final Subscriber&lt;? super T&gt; a = downstream;
    final SimpleQueue&lt;T&gt; q = queue;
    // ... 생략 ...
    for (;;) {
        try {
            v = q.poll();
        } catch (Throwable ex) {
            // ... 생략 ...
            a.onError(ex); // NOTE
            worker.dispose();
            return;
        }

        boolean empty = v == null;

        // NOTE: checkTerminated() 반환값이 true인 경우, 반환하기 전에 함수 내에서 a.onError() 또는 a.onComplete()실행됨
        if (checkTerminated(d, empty, a)) {
            return;
        }
        // ... 생략 ...
        a.onNext(v); // NOTE
    }
}
</code></pre>

<p>무한 반복문을 돌면서, <code>queue</code>에 있는 value들을 하나씩 처리하면서 적절한 downstream <code>Subscriber</code>의 함수를 실행하게 된다. <br>
즉, downstream <code>Subscriber</code>의 함수들은 설정한 <code>Schedule</code>의 <code>Worker</code>에 의해 스케쥴링된다.</p>

<hr>

<h3 id="">정리</h3>

<p>구현에서 중요한 부분만 최대한 요약할려고 했음에도, 설명이 다소 길어졌던 것 같다. 구현 부분에 대한 설명이 길었지만, 확인한 결과를 간단히 요약해서 정리하면(<code>Flowable</code> 기준),</p>

<ul>
<li><code>subscribeOn(Scheduler scheduler)</code>: upstream <code>Subscription</code> 의 <code>reqeust()</code>를 실행할 <code>Scheduler</code> 를 설정</li>
<li><code>observeOn(Scheduler scheduler)</code>: downstream <code>Subscriber</code> 의 동작(<code>onNext()</code>, <code>onError()</code>, <code>onComplete()</code>) 을 실행할 <code>Scheduler</code>를 설정</li>
</ul>

<p>이 2가지만 기억한다면, 하나 이상의 <code>subscribeOn()</code> 또는 <code>observeOn()</code>을 붙였을 때 일어나는 스케쥴링 동작들을 이해하는데 큰 도움이 될 것이다.   </p>

<p>아래는 추가로 Reactive Streams의 규격을 따르고 있지는 않지만, Rxjava2의 <code>Single</code>, <code>Maybe</code>에서의 동작에 대해서도 적어보았다. <br>
Reactive Streams의 <code>Publiser</code>, <code>Subscriber</code>가 각각 <code>SingleSource</code>(또는 <code>MaybeSource</code>), <code>SingleObserver</code>(또는 <code>MaybeObserver</code>)에 대응된다고 생각하면 <code>Flowable</code>과 거의 유사하다.</p>

<ul>
<li><code>subscribeOn(Scheduler scheduler)</code>: upstream <code>SingleSource</code>(또는 <code>MaybeSource</code>)의 <code>subscribe()</code>를 실행할 <code>Scheduler</code> 를 설정</li>
<li><code>observeOn(Scheduler scheduler)</code>: downstream <code>SingleObserver</code>(또는 <code>MaybeObserver</code>)의 동작(<code>onSuccess()</code>, <code>onError()</code>, <code>onComplete()</code>) 을 실행할 <code>Scheduler</code>를 설정</li>
</ul>

<p><strong><a id="jupiny-blog-link" href="https://jupiny.com/2020/04/22/rxjava-subscribeon-observeon-2/" target="_blank">다음 글</a></strong> 에서는 실제 로그를 통해 이 동작을 확인해보도록 할 것이다.</p>

<hr>

<h3 id="">참고</h3>

<ul>
<li><strong><a id="jupiny-blog-link" href="https://engineering.linecorp.com/ko/blog/reactive-streams-with-armeria-1/" target="_blank">Armeria로 Reactive Streams와 놀자! – 1</a></strong></li>
</ul>
        </section></div>

jupiny의 개발일지

현재 약 2년동안 실무에 RxJava를 사용하고 있지만, 아직까지(부끄럽게도) 항상 동작이 헷갈리는 함수 2개가 있다. 
subscribeOn() 과 observeOn()이다. 이미 많은 블로그들의 글을 읽어봤지만, 여전히 헷갈리는 듯하여 RxJava 내부적인 구현의 관점에서 나름대로 한번 정리를 해볼려고 한다.
Reactive Streams
먼저 RxJava의 subscribeOn() 과 observeOn()의 동작을 이해하기 위해서는, Reactive Streams 의 기본 메커니즘을 이해하여야 한다. RxJava도 결국 이 Reactive Streams의 규격을 따르는 구현체의 하나일 뿐이기 때문이다.
API Specification
Reactive Streams API 명세를 살펴보면, 아래와 같이 3개의 interface Publisher, Subscription, Subscriber 가 있다. (사실 Processor도 있지만, 여기서는 무시)
public interface Publisher<T> {  
    // 새로운 Subscription을 하나 만들어 Subscribe의 onSubscribe() 호출 
    public void subscribe(Subscriber<? super T> s);
}
public interface Subscription {  
    // Publisher에게 n개의 데이터 발행 요청
    public void request(long n);
    // Publisher에게 데이터 발행 중단 요청
    public void cancel();
}
public interface Subscriber<T> {  
    // Subscription의 request() 호출
    // Publisher는 그에 대한 응답을 내려줌
    public void onSubscribe(Subscription s);
    // Subscription의 request() 요청에 대한 Publisher의 데이터 응답
    public void onNext(T t)
    // Publisher의 데이터 발행 중의 에러 공지
    public void onError(Throwable t);
    // Publisher의 데이터 발행 완료 공지
    public void onComplete();
}
Flow
Subscriber 생성하여 Publisher의 subscribe(Subscriber s) 실행  
Publisher의 subscribe(Subscriber s): Subscription 생성하여 Subscriber의 onSubscribe(Subscription s) 실행  
Subscriber의 onSubscribe(Subscription s): Subscription의 request() 실행  
Subscription의 request(): Publisher에게 데이터 요청  
Publisher은 Subscription을 통해 Subscriber의 onNext()/onError()/onComplete() 실행
RxJava2의 Flowable 구현
코드를 통해 직접 확인해보기 위해 Reactive Streams의 규격을 잘 따르고 있는 RxJava2의 Flowable의 구현을 한번 살펴보자. 
Single, Mabye 의 경우는 Reactive Streams의 규격을 따르지 않고 있음.)
Rxjava2 2.2.19 버젼을 사용하였습니다.
subscribeOn
Flowable.just(1)  
        .subscribeOn(Schedulers.io()) // HERE
        .subcribe()
subscribeOn(Schedulers.io()) 코드의 구현을 따라가다 보면, SubscribeOnSubscriber이라는 FlowableSubscribeOn 클래스의 내부 클래스를 만나게 된다. 
@Override
public void onNext(T t) {  
    downstream.onNext(t); // NOTE
}
@Override
public void onError(Throwable t) {  
    downstream.onError(t); // NOTE
    worker.dispose();
}
@Override
public void onComplete() {  
    downstream.onComplete(); // NOTE
    worker.dispose();
}
일단 downstream의 Subscriber의 동작을 실행하는 부분에는 설정한 Scheduler로 별도로 스케줄링하지 않고 있다.
@Override
public void request(final long n) {  
    if (SubscriptionHelper.validate(n)) {
        Subscription s = this.upstream.get();
        if (s != null) {
            requestUpstream(n, s); // NOTE
        } else {
            // ... 생략 ...
        }
    }
}
void requestUpstream(final long n, final Subscription s) {  
    if (nonScheduledRequests || Thread.currentThread() == get()) {
        s.request(n);
    } else {
        worker.schedule(new Request(s, n)); // NOTE
    }
}
static final class Request implements Runnable {  
    final Subscription upstream;
    final long n;
    Request(Subscription s, long n) {
        this.upstream = s;
        this.n = n;
    }
    @Override
    public void run() {
        upstream.request(n); // NOTE
    }
}
request() 함수를 따라가보면, scheduler.createWorker()(위에 코드는 넣지 않았지만)로 생성한 Worker에 의해 upstream Subscription의 request()이 스케쥴링됨을 확인할 수 있다. 
observeOn
Flowable.just(1)  
        .observeOn(Schedulers.io()) // HERE
        .subcribe()
마찬가지로 observeOn(Schedulers.io()) 코드의 구현을 따라가다 보면, ObserveOnSubscriber이라는 FlowableObserveOn 클래스의 내부 클래스를 만나게 된다.  그리고 이 ObserveOnSubscriber 상속하고 있는 추상 클래스 BaseObserveOnSubscriber 를 살펴보면,
@Override public final void onNext(T t) {
    // ... 생략 ...
    trySchedule(); // NOTE
}
@Override
public final void onError(Throwable t) {  
    // ... 생략 ...
    trySchedule(); // NOTE
}
@Override
public final void onComplete() {  
    // ... 생략 ...
    trySchedule(); // NOTE
}
Subscriber 모든 함수의 마지막에 trySchedule()가 실행됨을 볼 수 있다. trySchedule() 의 구현을 보면,
final void trySchedule() {  
    if (getAndIncrement() != 0) {
        return;
    }
    worker.schedule(this); // NOTE
}
역시 scheduler.createWorker()를 통해 생성된 Worker를 통해 실행되는데, 여기서 this는 Runnable 를 구현한 자신이다. 
Runnable를 구현한 부분을 보면,
@Override
public final void run()  
    // ... 생략 ...
    runAsync(); // NOTE
}
runAsync()이라는 추상 메쏘드를 실행한다. 처음으로 돌아와서 이 추상 클래스를 상속한 ObserveOnSubscriber에서 구현한 runAsync() 을 확인해보면,
@Override
void runAsync() {  
    // ... 생략 ...
    final Subscriber<? super T> a = downstream;
    final SimpleQueue<T> q = queue;
    // ... 생략 ...
    for (;;) {
        try {
            v = q.poll();
        } catch (Throwable ex) {
            // ... 생략 ...
            a.onError(ex); // NOTE
            worker.dispose();
            return;
        }
        boolean empty = v == null;
        // NOTE: checkTerminated() 반환값이 true인 경우, 반환하기 전에 함수 내에서 a.onError() 또는 a.onComplete()실행됨
        if (checkTerminated(d, empty, a)) {
            return;
        }
        // ... 생략 ...
        a.onNext(v); // NOTE
    }
}
무한 반복문을 돌면서, queue에 있는 value들을 하나씩 처리하면서 적절한 downstream Subscriber의 함수를 실행하게 된다. 
Subscriber의 함수들은 설정한 Schedule의 Worker에 의해 스케쥴링된다.
정리
구현에서 중요한 부분만 최대한 요약할려고 했음에도, 설명이 다소 길어졌던 것 같다. 구현 부분에 대한 설명이 길었지만, 확인한 결과를 간단히 요약해서 정리하면(Flowable 기준),
subscribeOn(Scheduler scheduler): upstream Subscription 의 reqeust()를 실행할 Scheduler 를 설정
observeOn(Scheduler scheduler): downstream Subscriber 의 동작(onNext(), onError(), onComplete()) 을 실행할 Scheduler를 설정
이 2가지만 기억한다면, 하나 이상의 subscribeOn() 또는 observeOn()을 붙였을 때 일어나는 스케쥴링 동작들을 이해하는데 큰 도움이 될 것이다.   
아래는 추가로 Reactive Streams의 규격을 따르고 있지는 않지만, Rxjava2의 Single, Maybe에서의 동작에 대해서도 적어보았다. 
Publiser, Subscriber가 각각 SingleSource(또는 MaybeSource), SingleObserver(또는 MaybeObserver)에 대응된다고 생각하면 Flowable과 거의 유사하다.
subscribeOn(Scheduler scheduler): upstream SingleSource(또는 MaybeSource)의 subscribe()를 실행할 Scheduler 를 설정
observeOn(Scheduler scheduler): downstream SingleObserver(또는 MaybeObserver)의 동작(onSuccess(), onError(), onComplete()) 을 실행할 Scheduler를 설정
다음 글 에서는 실제 로그를 통해 이 동작을 확인해보도록 할 것이다.
참고
Armeria로 Reactive Streams와 놀자! – 1

RxJava의 subscribeOn와 observeOn (1)

댓글