rxjava 가 메이저 버전 업(1->2)을 하면서 몇 가지 변경점이 생겼다.

변경점에 대한 자세한 내용은 아래 링크를 참조하기 바란다.

• rxJava Wiki : https://github.com/ReactiveX/RxJava/wiki/What’s-different-in-2.0
• 번역 : http://realignist.me/code/2017/01/25/rxjava2-changelog.html

Flowable 이라는 base reactive class 가 추가 되었다. Observable 과의 차이는 backpressure buffer의 기본 탑재 유무이다.

backpressure?

우리말로 번역하면 ‘등 뒤에서 떠밀리는 압박’ 정도가 될 듯 하다.

이런 상황을 가정해보자. 콘서트장을 사람들이 가득 메웠다. 콘서트장에 들어오려는 사람들은 저글링 개떼처럼 밀려드는데 나가는 사람은 별로 없다. 콘서트장 출입구를 통제하는 요원이 없다면? 콘서트장이 터지던지 안에 있던 사람들이 짜부러지던지 아무튼 대형 사고가 발생할거다.

publish / subscribe 모델에서도 이런 비극적인 시나리오가 발생할 수 있다. 생산자는 미친듯이 element 를 생산해 내는데 소비자가 처리하는 속도가 이를 따라가지 못한다면

1. busy waiting 또는
2. out of memory exception 이 발생할 것이다.

‘등 뒤에서 떠밀리는 압박’ 에 대한 흐름제어를 위한 버퍼가 바로 backpressure buffer 다. 버퍼가 가득 차면 어차피 소비자는 element 를 처리할 여유가 없는 상태이므로 더 이상 publish 를 하지 않는다.

기존에 없던 개념이 새로 추가된 것은 아니다. 기존 rxJava 1.xx 의 경우 Observable 에 backpressure buffer 를 직접 생성해 주면 사용이 가능하다. 허나 rxJava 개발자는 초보자들이 미처 알아채지 못하는 영역에서 기대하지 않는 동작이 일어날 가능성이 있다며 Flowable 을 추가하였다.

다음 예제코드를 보자. 생산자의 생산 속도를 소비자가 따라가지 못하는 시나리오다.
Flowable 을 사용하면 default buffer size(128) 이상 backpressure buffer 에 element 가 쌓일 경우 흐름제어를 한다.

public class example01 {

    public static void main(String... args) throws InterruptedException {

        final String tmpStr = Arrays.stream(new String[10_000_000]).map(x->"*").collect(Collectors.joining());
        Flowable foo = Flowable.range(0, 1000_000_000)
                .map(x-> {
                    System.out.println("[very fast sender] i'm fast. very fast.");
                    System.out.println(String.format("sending id: %s %d%50.50s", Thread.currentThread().getName(), x, tmpStr));
                    return x+tmpStr;
                });

        foo.observeOn(Schedulers.computation()).subscribe(x->{
            Thread.sleep(1000);
            System.out.println("[very busy receiver] i'm busy. very busy.");
            System.out.println(String.format("receiving id: %s %50.50s", Thread.currentThread().getName(), x));
        });

        while (true) {
            Thread.sleep(1000);
        }
    }
}

[very fast sender] i'm fast. very fast.
sending id: main 0**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 1**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 2**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 3**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 4**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 5**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 6**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 7**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 8**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 9**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 10**************************************************

... 중략 ...

[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 0*************************************************
receiving id: RxComputationThreadPool-1 1*************************************************
receiving id: RxComputationThreadPool-1 2*************************************************
[very fast sender] i'm fast. very fast.
sending id: main 117**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 118**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 119**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 120**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 121**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 122**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 123**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 124**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 125**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 126**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 127**************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 3*************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 4*************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 5*************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 6*************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 7*************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 8*************************************************
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 9*************************************************

반면, 같은 시나리오를 Observable 을 backpressure buffer 생성 없이 사용하면 OutOfMemoryException 이 발생한다.

public class example02 {

    public static void main(String... args) throws InterruptedException {

        final String tmpStr = Arrays.stream(new String[10_000_000]).map(x->"*").collect(Collectors.joining());
        Observable foo = Observable.range(0, 1000_000_000)
                .map(x-> {
                    System.out.println("[very fast sender] i'm fast. very fast.");
                    System.out.println(String.format("sending id: %s %d%50.50s", Thread.currentThread().getName(), x, tmpStr));
                    return x+tmpStr;
                });

        foo.observeOn(Schedulers.computation()).subscribe(x->{
            Thread.sleep(1000);
            System.out.println("[very busy receiver] i'm busy. very busy.");
            System.out.println(String.format("receiving id: %s %50.50s", Thread.currentThread().getName(), x));
        });

        while (true) {
            Thread.sleep(1000);
        }
    }
}

[very fast sender] i'm fast. very fast.
sending id: main 0**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 1**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 2**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 3**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 4**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 5**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 6**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 7**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 8**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 9**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 10**************************************************
[very fast sender] i'm fast. very fast.

...중략...

sending id: main 198**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 199**************************************************
[very fast sender] i'm fast. very fast.
sending id: main 200**************************************************
Exception in thread "main" java.lang.OutOfMemoryError: Java heap space
	at java.util.Arrays.copyOfRange(Arrays.java:3664)
	at java.lang.String.<init>(String.java:207)
	at java.lang.StringBuilder.toString(StringBuilder.java:407)
	at example02.lambda$main$1(example02.java:24)
	at example02$$Lambda$6/123961122.apply(Unknown Source)
	at io.reactivex.internal.operators.observable.ObservableMap$MapObserver.onNext(ObservableMap.java:59)
	at io.reactivex.internal.operators.observable.ObservableRange$RangeDisposable.run(ObservableRange.java:64)
	at io.reactivex.internal.operators.observable.ObservableRange.subscribeActual(ObservableRange.java:35)
	at io.reactivex.Observable.subscribe(Observable.java:10700)
	at io.reactivex.internal.operators.observable.ObservableMap.subscribeActual(ObservableMap.java:33)
	at io.reactivex.Observable.subscribe(Observable.java:10700)
	at io.reactivex.internal.operators.observable.ObservableObserveOn.subscribeActual(ObservableObserveOn.java:45)
	at io.reactivex.Observable.subscribe(Observable.java:10700)
	at io.reactivex.Observable.subscribe(Observable.java:10686)
	at io.reactivex.Observable.subscribe(Observable.java:10589)
	at example02.main(example02.java:27)
[very busy receiver] i'm busy. very busy.
receiving id: RxComputationThreadPool-1 5*************************************************

참고로, Flowable 은 FlowableCreate 라는 builder 에서 생성되며, 특별한 설정이 없을 경우 buffer size 는 최소 16, 기본 128 로 설정한다.

//FlowableCreate.java line:44

    @Override
    public void subscribeActual(Subscriber<? super T> t) {
        BaseEmitter<T> emitter;

        switch (backpressure) {
        case MISSING: {
            emitter = new MissingEmitter<T>(t);
            break;
        }
        case ERROR: {
            emitter = new ErrorAsyncEmitter<T>(t);
            break;
        }
        case DROP: {
            emitter = new DropAsyncEmitter<T>(t);
            break;
        }
        case LATEST: {
            emitter = new LatestAsyncEmitter<T>(t);
            break;
        }
        default: {
            emitter = new BufferAsyncEmitter<T>(t, bufferSize());
            break;
        }
        }

        t.onSubscribe(emitter);
        try {
            source.subscribe(emitter);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            emitter.onError(ex);
        }
    }

//Flowable.java line:61
    static final int BUFFER_SIZE;
    static {
        BUFFER_SIZE = Math.max(16, Integer.getInteger("rx2.buffer-size", 128));
    }

참고 문서

개요

작년에 wildcard에 대해 스터디한 내용이 있었다는 사실에  한 번 개깜놀

까맣게 잊고 있었다는 사실에 두 번 개깜놀

wildcard 와 type parameter 간의 가장 중요한 차이점은 code statement 내에 선언된 param이 재참조되는지 여부이다.

재참조 되면 type parameter 써야 하고, 재참조 안되고 귀찮으면 그냥 ? 쓴다.

Wildcard ?

java syntax에는 ? 라는 것이 있다. generic type parameter로 쓰인다. (e.g. ArrayList<?>)

Long story short…

generic type parameter 를 명확히 결정하지 않은 채로 두기 위한(=unknown type) syntax

Generic 의 type parameter 는 polymorphism 을 지원하지 않는다.

generic에서는 type prarmeter 에 대하여 polymorphism 을 지원하지 않는다.  다음 예제를 보자.

Generic

Collection<String> foo = new ArrayList<String>;
Collection<Object> refFoo = foo; //compile error!!

refFoo.add(new Object());
String bar = foo.get(0); //만일 line2 가 error가 아니었다면, bar 에 Object 를 할당하는 참사가 발생하게 된다.

얼핏 생각하면 ArrayList<Dog> 을 List<Animal> 으로 받을 수 있을 것 같지만, 상기 예제와 같은 문제를 막기 위해 Generic 의 type parameter 는 문법적으로 subtype 변환을 허용하지 않는다.

혼란을 막기 위한 제한이지만, type 체크가 엄격하면 유연한 사용이 힘들다. Wildcard 는 이런 상황을 해결하기 위해 고안된 문법이다.

private void foo(Collection<?> c) {
  for (Object iter : c)
    System.out.println(iter);
}

List<String> bar = new ArrayList<String>;
bar.add("hello");

foo(bar); //okay!!

private void foo(Collection<Object> c) {
  for (Object iter : c)
    System.out.println(iter);
}

List<String> bar = new ArrayList<String>;
bar.add("hello");

foo(bar); //compile error!!

Wildcard 사용시 주의할 점

bounded wildcard

spring에서 용례

spring-context 의 SimpleApplicationEventMulticaster.multicastEvent()

정확한 type을 알 수 없는 ApplicationListner<T>를 이터레이션 하기 위하여 wildcard가 사용되었다.

SimpleApplicationEventMulticaster.java

@Override
public void multicastEvent(final ApplicationEvent event) {
  for (final ApplicationListener<?> listener : getApplicationListeners(event)) {
    Executor executor = getTaskExecutor();
    if (executor != null) {
      executor.execute(new Runnable() {
        @Override
        public void run() {
          invokeListener(listener, event);
        }
      });
    }
    else {
      invokeListener(listener, event);
    }
  }
}