박철우의 블로그

Java 11

새로운 String 메서드 추가

- strip(): 문자열 앞, 뒤의 공백 제거.

- stripLeading(): 문자열 앞의 공백 제거.

- stripTrailing(): 문자열 뒤의 공백 제거.

trim() 과의 차이점은, trim() 은 U+0020 이하의 값만을 공백으로 인식하여 제거한다 (tab, CR, LF, 공백). 하지만 유니코드에서는 이 외에 다양한 공백 문자가 존재하는데, 이를 처리하기 위해서는 기존에는 Character.isWhitespace(int) 를 사용해야 했다. Java 11 부터는 strip() 으로 편하게 처리할 수 있다.

그리고 성능도 수 배 빠른 것으로 알려짐.

- isBlank(): 문자열이 비어있거나 공백만 포함되어 있을 경우 true 를 반환한다. 즉, String.trim().isEmpty() 호출 결과와 같다.

- lines(): 문자열을 라인 단위로 쪼개는 스트림을 반환.

- repeat(n): 지정된 수 만큼 문자열을 반복하여 붙여 반환:

String str = "ABC";
String repeated = str.repeat(3);	// "ABCABCABC"

java.nio.file.Files 클래스 유틸 메서드 추가

- Path writeString(Path, String, Charset, OpenOption): 파일에 문자열을 작성하고 Path로 반환한다. 파일 오픈 옵션에 따라 작동 방식을 달리하며, charset을 지정하지 않으면 utf-8 이 사용된다. (오버로딩 메서드로 writeString(Path, String, OpenOption) 존재.)

- String readString(Path, Charset): 파일 전체 내용을 읽어서 String으로 반환하고, 파일 내용을 모두 읽거나 예외가 발생하면 알아서 close 한다. charset을 지정하지 않으면 utf-8 이 사용된다. (오버로딩 메서드로 readString(Path) 존재.)

- boolean isSameFile(Path, Path): 두 Path 가 같은 파일을 가리키면 true, 아니면 false 를 반환한다. 파일이 실제로 존재하지 않아도, Path 를 기준으로 해서 같은 위치면 true 로 판단한다.

Pattern.asMatchPredicate()

- Java 8 의 asPredicate는 matcher().find() 를 사용하는 것에 반해, asMatchPredicate() 는 matcher().match() 를 사용하는 Predicate를 반환한다.

Predicate.not(Predicate)

- 인자로 받은 Predicate의 부정형 Predicate를 반환한다.

람다 파라미터로 var 사용:

(var n1, var n2) -> n1 + n2

- Java 8 에 등장했으나 Java 10 에서 사라졌다가 Java 11 에서 복귀한 기능.

- 람다는 타입을 스킵할 수 있는데 이걸 사용하는 이유는, @Nullable 등의 어노테이션을 사용하기 위해 타입을 명시해야 할 때.

- var 를 사용하려면 괄호를 써야하며, 모든 파라미터에 사용해야 하고, 다른 타입과 혼용하거나 일부 스킵은 불가능함:

(var s1, s2) -> s1 + s2 // 안 됨. s2 에도 var 필요.
(var s1, String y) -> s1 + y // 안 됨. String 과 혼용 불가.

var s1 -> s1 // 안 됨. 괄호 필요.

Optional.isEmpty()

- Optional이 비어있을 때 true 반환.

TimeUnit.convert(Duration)

TimeUnit c = TimeUnit.DAYS;
c.convert(Duration.ofHours(24));	// 2
c.convert(Duration.ofHours(72));	// 3

Nest-Based Access Control (JEP 181)

- Java 10 까지 nested 클래스에서 자신, 혹은 outer 클래스의 private 멤버에 리플렉션을 통한 접근을 시도하면 IllegalAccessException 이 발생했고, 이를 피하기 위해 setAccessible(true) 을 호출해야만 했음.

- Java 11 부터는 setAccessible 없이 private 멤버를 호출할 수 있음.

- java.lang.Class 에 다음 메서드들이 추가됨.

- getNestHost(): nested 클래스에서 호출하면 자신을 감싸고 있는 outer 클래스를 반환하고, outer 클래스에서 호출하면 자신을 반환.

- boolean isNestmateOf(Class): 자신이 아규먼트로 받은 Class의 nested 클래스일 때 true 를 반환함.

- Class[] getNestMembers(): 자신을 포함하여 중첩 관계에 있는 모든 클래스를 배열로 반환함. (outer, siblings)

Epsilon Garbege Collector (No-Op GC, JEP 318)

- JVM으로 하여금 메모리 할당을 관리하지만, 사용된 메모리를 재사용하지 않도록 함. 메모리를 다 사용하면 OutOfMemory가 발생하고 JVM은 셧다운된다.

- 어플리케이션 테스트에 사용됨. GC는 어플리케이션과 함께 작동하며, 오버헤드가 있어 어플리케이션 성능에 영향을 준다. No-OP GC를 적용하여 GC를 배제함으로써 순수 어플리케이션의 성능, 메모리 부하 등을 테스트할 수 있도록 한다. GC 적용 시의 성능과 비교하여 GC의 영향도를 측정할 수 있다.

- 짧은 시간 수행하고 종료되는 어플리케이션에 사용됨. 간단히 작동하고 마치는, 메모리 부하가 크게 염려되지 않는 어플리케이션에 적절하게 사용될 수 있음.

- 아래 아규먼트를 사용하여 활성화한다:

-XX:+UnlockExperimentalVMOptions -XX:+UseEpsilonGC

Dynamic Class-File Constants (JEP 309)

Java Flight Recorder  (JEP 328)

- 상업용 OracleJDK 에 제공되던 JFR이 OpenJDF에 포함된다.

- Java 어플리케이션으로부터 프로파일링, 어플리케이션 진단 데이터 등을 얻을 수 있음.

- 성능 오버헤드가 1% 미만으로 알려져 있어, 운영 환경에서 사용할 수 있음.

HTTP Client (JEP 321)

- Java 표준 HTTP 클라이언트 API.

- 그간 HTTP 통신을 위해 사용된 코드보다 성능이 개선됨.

- HTTP/1.1, HTTP/2, WebSocket을 지원한다. 

java 파일 실행 (JEP 330)

- javac를 통한 컴파일 없이 java 파일을 실행할 수 있음:

$> java HelloWorld
Hello world!!

- java 파일에 main 메서드가 존재해야 함.

- 다음과 같이 클래스패스 지정:

java --class-path=/myclasspath ExecutionTest.java

- 클래스패스에 동일한 이름의 클래스가 존재하면 에러 발생. (java 확장자 무시.)

기타 제거된 기능 및 옵션

com.sun.awt.AWTUtilities 클래스

Lucida 폰트 (Oracle JDK)

appletviewer Launcher

javax.imageio JPEG 플러그인 alpha 이미지 지원 X

sun.misc.Unsafe.defineClass

Thread.destroy()

Thread.stop(Throwable)

JVM-MANAGEMENT-MIB.mib

SNMP 에이전트

JavaFX (Oracle JDK)

Java EE (JAX-WS, JAXB, JAF, Common Anotations) (JEP 320)

CORBA (JEP 320)

deprecated된 기능 및 옵션

ThreadPoolExecutor는 Finalization에 의존성을 갖지 않도록 (should not)

Nashorn 자바스크립트 엔진 (JEP 335)

-XX+AggressiveOpts

Pack200 툴 및 API (JEP 336)

스트림 기반 GSSContext 메서드

출처:

https://dzone.com/articles/features-of-java-11

https://www.journaldev.com/24601/java-11-features

https://www.geeksforgeeks.org/java-11-features-and-comparison/

Java 10

var 키워드

- 지역변수에만 사용 가능.

- 컴파일 타임에 타입이 확실한 곳에만 사용 가능.

- 컴파일 타임에 타입 추정(type inference)이 적용되어, 바이트코드는 이전과 동일하게 생성.

Unmodifiable View Collections

- 컬렉션을 wrapping하여 변경 불가능한 컬렉션을 만든다.

- immutable 과는 다름. 직접 데이터를 가지고 있지 않고, 원본 인스턴스의 view 로서 작동한다. 원본 인스턴스에 요소가 추가/삭제/변경되면 이를 래핑한 unmodifiable collections 도 변경되어 보여진다. 글자 그대로 VIEW.

List unmodifiable = Collections.unmodifiableList(existingList);

Creating Immutable Collections 

- List, Set, Map 에 copyOf(...) 등장. 얉은 복사를 수행하여 immutable collection을 반환한다.

List immutableCopied = List.copyOf(existingList);

- 공간 효율성이 다르다. 아래 코드에서, HashSet 은 내부의 HashMap 을 포함하여 6개의 인스턴스를 가진다:

Set<String> set = new HashSet<>(3);   // 3 buckets
set.add("silly");
set.add("string");
set = Collections.unmodifiableSet(set);

하지만 아래 코드는 2개 필드를 가진 인스턴스를 1개 생성한다. heap 사용률을 줄일 수 있다:

Set<String> set = Set.of("silly", "string");

Immutable Collectors

- 스트림의 컬렉터로 immutable collection 을 생성하는 컬렉터가 추가됨:

   Set<Item> immutableSet =
      sourceCollection.stream()
                      .map(...) 
                      .collect(Collectors.toUnmodifiableSet());

G1GC 개선

- Java 9 에서 디폴트 GC 알고리즘으로 채택된 G1GC 의 full gc 방식을 개선.

- Full GC 시 전체 프로세싱을 병렬로 수행하여 성능 향상.

CDS (Class Data Sharing) 옵션 추가

- Java 5 에서 도입된 기존 CDS 는 부트스트랩 클래스로더의 대상 클래스만을 쉐어링한다.

- 다음 아규먼트를 통해 시스템 클래스로더와 플랫폼 클래스로더로 쉐어링하도록 설정 가능:

-XX:+UseAppCDS

- 어플리케이션 클래스를 쉐어링하도록 설정할 수 있음:

Step 1. 쉐어링 대상 클래스 지정:

java -Xshare:off -XX:+UseAppCDS -XX:DumpLoadedClassList=myapp.lst \
  -cp $CLASSPATH $MAIN_CLASS

Step 2. CDS 아카이브 생성:

java -Xshare:dump -XX:+UseAppCDS -XX:SharedClassListFile=myapp.lst \
  -XX:SharedArchiveFile=myapp.jsa \
  -cp $CLASSPATH

Step 3. 어플리케이션 실행:

java -Xshare:on -XX:+UseAppCDS -XX:SharedArchiveFile=hello.jsa \
    -cp $CLASSPATH $MAIN_CLASS

JIT 컴파일러 관련

- 기존 JIT 컴파일러는 C++ 로 작성되어, 변경에 어려움이 있음

- 순수 자바로 JIT 컴파일러를 작성 가능하도록 하는 JVM Compiler Interface (JVMCI) 등장.

- 그 결과로, 순수 자바로 작성된 Graal 등장.

- 아직 실험적인 단계여서, 다음 아규먼트를 설정할 때만 작동:

-XX:+UnlockExperimentalVMOptions -XX:+UseJVMCICompiler

Thread-Local 핸드셰이크

- 기존 JVM은 모든 쓰레드의 스택트레이스를 얻거나 GC를 수행하기 위해서는 한 모든 쓰레드를 정지시켜야 했음(STW)

- STW 의 이유는, JVM이 작업을 마친 뒤 쓰레드들이 하던 작업을 그대로 다시 수행하도록 하기 위해 글로벌 세이프포인트를 생성하기 위함. 세이프포인트가 있어야 쓰레드가 다시 작동할 컨텍스트를 알 수 있고, 그래야 쓰레드를 멈출 수 있음.

- Java 10 부터 JVM은 임의의 숫자의 쓰레드 세이프포인트를 생성할 수 있음. 고로, 한 번에 한 쓰레드만 멈출 수 있게 됨.

컨테이너 인식

- 전까진 JVM은, CPU, 메모리 등의 자원을 호스트 OS 에서 쿼리했음.

- Java 10 부터는 JVM이 도커 컨테이너 안에서 작동하는 경우, 도커 컨테이너에 할당된 자원을 쿼리할 수 있게 됨.

- 예로, Java 9 까지는 호스트 OS 의 가용 메모리가 2G 이고, JVM이 작동하는 컨테이너에는 0.5G 만 할당되어 있는 환경에서 Java는 메모리를 2G 로 쿼리한다. Java 10 부터는 도커의 컨테이너 컨트롤 그룹에서 자원을 쿼리할 수 있도록 하여 컨테이너 환경에서의 실제 가용 자원을 정확하게 얻을 수 있음.

- 컨테이너 환경에서 힙메모리와 프로세서 수를 지정하는 아규먼트가 존재함:

-XX:+UseCGroupMemoryLimitForHeap 
-XX:ActiveProcessorCount=2

OpenJDK 의 SSL 관련

출처: 

https://dzone.com/articles/whats-new-in-java-10

Point 1. subscribeOn 은 업스트림과 다운스트림에, observeOn 은 다운스트림에 영향을 준다는 점을 기억한다.

Point 2. 하나의 파이프라인에는 하나의 쓰레드만 작동한다. (flatMap 을 통한 병렬화를 하지 않을 경우.)

Point 3. 워커 쓰레드가 요소를 처리 중일 때 다음 요소가 들어오면 이 요소는 queuing되고, 현재 요소의 처리가 끝나면 queuing된 다음 요소를 처리한다.

Point 4. 처리 지연에 대비하여 별도 스케쥴러 워커에 실행시킬 로직이 있고, observeOn 을 사용하려 한다면 이 로직은 반드시 observeOn 다음에 위치시켜야 한다.

스케쥴러 쓰레드 기아 현상은 다음과 같은 경우 발생한다:

// 아래 두 source는 1초에 한 번 Item을 발행하는 Hot Observable 이라 가정한다.
Observable<Item> source1 = ... // observeOn 스케쥴링이 걸린 소스 1 (워커A)
Observable<Item> source2 = ... // observeOn 스케쥴링이 걸린 소스 2 (워커B)

Observable.merge(source1, source2)
		.subscribe(item -> {
        	... // 1초 넘는 지연 발생
		});

위 코드에서 작업을 수행하는 쓰레드는 워커A, 워커B 두 쓰레드이다. observeOn은 다운스트림으로 subscribe까지 영향을 미치므로, 위 코드에서 merge 의 subscribe는 Item이 방출된 워커 쓰레드가 수행하려 한다. (source1에서 나온 Item은 워커A가, source2에서 나온 Item은 워커B가.)

그런데 subscribe에서 지연이 발생할 경우 이야기는 달라지고, 위 코드는 다음과 같은 결과를 낳는다. (워커의 순서는 무관하며, Item 구분을 위해 워커A의 아이템은 ItemA-{순서}, 워커B의 아이템은 ItemB-{순서} 라 칭한다.)

1. 워커A가 ItemA-1을 방출하고 subscribe 안의 무거운 로직을 처리하러 들어간다.

2. 워커B가 ItemB-1을 방출하고 subscribe 으로 Item을 보낸다. 이 때 워커A가 이미 ItemA-1을 처리하는 중이기 때문에 ItemB-1은 queuing된다.

3. 워커A는 ItemA-1의 처리를 마치고 queuing되었던 ItemB-1 처리를 시작한다.

4. 워커B가 ItemB-2을 방출한다. 워커A는 워커B의 첫 번째 ItemB-1을 처리하는 중이기 때문에 이 ItemB-2는 queuing된다.

5. 워커A는 ItemB-1의 처리를 마치고 ItemB-2 처리를 시작한다.

6. 워커B는 ItemB-3를 방출한다. 워커A는 ItemB-2를 처리하는 중이기 때문에 ItemB-3는 queuing된다.

7. 워커A는 ItemB-2의 처리를 마치고 ItemB-3 처리를 시작한다.

6. 워커B는 ItemB-4를 방출한다. 워커A는 ItemB-3를 처리하는 중이기 때문에 ItemB-4는 queuing된다.

...

위와 같이 워커A는 merged Observable로 queuing되는 Item을 처리하기 위해 계속 subscribe안에 머물게 된다. 문제는 source Observable이 Hot Observable이라는 것이다. 워커B가 ItemB-*를 계속해서 내보내고 queuing시키는 동안 ItemA-*또한 발생할 수 있는데, 워커A가 merged Observable에서 ItemB-*를 처리하느라 ItemA-*를 핸들링하지 못한다. merged Observable에 먼저 진입한 워커 쓰레드는 갇히게 되는 것이다.

Hot Observable이 아니더라도 이 현상은 똑같이 발생할 수 있는데, Cold Observable의 경우 모든 ItemB-* 처리를 마치고 나서야 ItemA-* 핸들링을 시작하게 된다. Hot Observable은 무한히 갇히게 된다는 점, Cold Observable은 마치 concatated Observable (Observable.concat(...)) 처럼, 하나의 source를 완전히 마치기 전까지 다른 source를 처리하지 못한다. 이는 merge 를 사용한 의도에 부합되지 않는다.

이 현상을 방지하기 위해서는 merged Observable의 로직 처리 또한 다른 워커 쓰레드에게 맡겨야 한다. 그 방법은 다양하겠지만, 대충 아래와 같이 실현할 수 있다:

...
Observable.merge(source1, source2)
		.observeOn(Schedulers.io())
		.subscribe(item -> {
        	... // 1초 넘는 지연 발생
		});

ConnectableObservable을 사용하여 One publicher-Many subscribers를 구현할 때, onError* 사용 시 subscriber마다 서로 다른 에러 처리 로직을 사용할 수 있다.

이 때, publisher 단계에서 에러가 발생하면 원치 않는 Subscriber의 에러 처리 코드가 실행될 수 있음을 유의하여야 한다.

아래와 같이 한 Publisher에 세 개의 Subscriber가 있을 때, observer 마다 서로 다른 에러 처리 로직을 가지고 있고, 자신만의 방법으로 에러를 처리하기를 원할 수 있다

source (ConnectableObservable) <- observer1, observer2, observer3

이 때 source 에서 에러가 발생하면 이 에러는 observer의 onError* 로 흐르게 된다. 이를 원치 않는다면, observer에서 flatMap-Observable.just를 사용하여 별도의 내부 파이프라인을 정의해야 한다.

RxJava2 에서, 하나의 Observable에는 하나의 쓰레드만 작동한다. 파이프라인에 스케쥴러를 사용해도 요소가 병렬로 처리되지는 않는다.

예로, 아래 코드를 보자:

...
Observable.create(new ObserveOnSubscribe<Item>() {
	@Override
    public void subscribe(ObservableEmitter<Item> emitter) throws Exception {
    	while (true) {
        	emitter.onNext(/* Item */);
            Thread.sleep(1000);
        }
    }
})
.observeOn(Schedulers.io())
.doOnNext(new Consumer<Item>() {
	@Override
    public void accept(Item item) throws Exception {
    	Thread.sleep(3000);    
    }
})
...

source는 1초에 한 번 Item을 발행한다. Schedulers.io()를 사용하였으므로 doOnNext에서 수행하는 로직이 1초를 초과해도 source의 Item 발행이 지연되지는 않는다.

이 때 사용자는 doOnNext의 수행이 1초를 초과하여 한 Item을 처리하기 전에 다음 Item이 발행되면 io 스케쥴러의 또다른 쓰레드가 작동하여 별도로 doOnNext로직을 수행할 거라고 생각할 수 있는데, RxJava2에서 한 Observable은 동시에 하나의 쓰레드만이 작동하기 때문에, doOnNext의 처리가 지연되어도 Item은 보이지 않게 계속 쌓이고, 현재 Item이 완전히 처리되어야 다음 Item을 처리한다.

위 코드에서 doOnNext는 3초의 처리 시간을 갖는다. 위 코드는 다음과 같은 흐름으로 실행된다.

Source 쓰레드] 1번 Item 발행
io 스케쥴러 쓰레드-1] 수행 시작: doOnNext(1번 아이템) 
--1초 경과--
Source 쓰레드] 2번 Item 발행
--1초 경과--
Source 쓰레드] 3번 Item 발행
--1초 경과--
io 스케쥴러 쓰레드-1] 수행 종료: doOnNext(1번 아이템)
io 스케쥴러 쓰레드-1] 수행 시작: doOnNext(2번 아이템) 
Source 쓰레드] 4번 Item 발행
--1초 경과--
Source 쓰레드] 5번 Item 발행
--1초 경과--
Source 쓰레드] 6번 Item 발행
--1초 경과--
io 스케쥴러 쓰레드-1] 수행 종료: doOnNext(2번 아이템)
io 스케쥴러 쓰레드-1] 수행 시작: doOnNext(3번 아이템) 
Source 쓰레드] 7번 Item 발행
--1초 경과--
Source 쓰레드] 8번 Item 발행
--1초 경과--
Source 쓰레드] 9번 Item 발행
...

물론 실제로 돌려보면 중간 경과 시간은 정확히 1초가 아닐 것이고, doOnNext 시작/종료와 Item 발행은 순서가 뒤바뀔 수 있다. 여튼 위처럼, 처리 쓰레드는 한 Item을 처리하는 데에 3초가 걸리고, Item은 1초에 하나씩 발행된다.

이처럼, observeOn은 처리 쓰레드를 달리 하는 것이지, 병렬 처리는 완전 별개의 문제다. Item 처리가 지연될 때 다른 쓰레드가 다음 Item을 처리하기를 원하면 다음과 같이 해야한다:

...
Observable.create(new ObserveOnSubscribe<Item>() {
	@Override
    public void subscribe(ObservableEmitter<Item> emitter) throws Exception {
    	while (true) {
        	emitter.onNext(/* Item */);
            Thread.sleep(1000);
        }
    }
})
.flatMap(new Function<Item, Observable<Item>>() {
	@Override
    public Observable<Item> apply(Item item) throws Exception {
    	return Observable.just(item)
			.observeOn(Schedulers.io())
            .doOnNext(new Consumer<Item>() {
                @Override
                public void accept(Item item) throws Exception {
                    Thread.sleep(3000);    
                }
            })
    }
})
...

Observable.just(item)을 통해 만들어진 Observable은 새로 생성된 별도의 Observable이고, 따라서 하나의 Observable에는 하나의 쓰레드만 작동한다는 성질이 별도로 적용된다. 이제 Item마다 새로운 Observable이 생성되므로 각자에 다른 io 스케쥴러 쓰레드가 할당된다.

위 코드의 실행 흐름은 다음과 같다:

Source 쓰레드] 1번 Item 발행
io 스케쥴러 쓰레드-1] 수행 시작: doOnNext(1번 아이템)
--1초 경과--
Source 쓰레드] 2번 Item 발행
io 스케쥴러 쓰레드-2] 수행 시작: doOnNext(2번 아이템)
--1초 경과--
Source 쓰레드] 3번 Item 발행
io 스케쥴러 쓰레드-3] 수행 시작: doOnNext(3번 아이템)
--1초 경과--
io 스케쥴러 쓰레드-1] 수행 종료: doOnNext(1번 아이템) 
Source 쓰레드] 4번 Item 발행
io 스케쥴러 쓰레드-1] 수행 시작: doOnNext(4번 아이템)
--1초 경과--
io 스케쥴러 쓰레드-2] 수행 종료: doOnNext(2번 아이템)
Source 쓰레드] 5번 Item 발행
io 스케쥴러 쓰레드-2] 수행 시작: doOnNext(5번 아이템)
--1초 경과--
io 스케쥴러 쓰레드-3] 수행 종료: doOnNext(3번 아이템)
Source 쓰레드] 6번 Item 발행
io 스케쥴러 쓰레드-3] 수행 시작: doOnNext(6번 아이템)
--1초 경과--
io 스케쥴러 쓰레드-1] 수행 종료: doOnNext(4번 아이템)
Source 쓰레드] 7번 Item 발행
io 스케쥴러 쓰레드-1] 수행 시작: doOnNext(7번 아이템)
--1초 경과--
io 스케쥴러 쓰레드-2] 수행 종료: doOnNext(5번 아이템)
Source 쓰레드] 8번 Item 발행
io 스케쥴러 쓰레드-2] 수행 시작: doOnNext(8번 아이템)
--1초 경과--
io 스케쥴러 쓰레드-3] 수행 종료: doOnNext(6번 아이템)
Source 쓰레드] 9번 Item 발행
io 스케쥴러 쓰레드-3] 수행 시작: doOnNext(9번 아이템)
...

Point 1. ConnectableObservable을 통해 하나의 source를 다수의 Observer가 구독할 수 있도록 한다.

Point 2. filter로 원하는 조건에 따라 source Observable을 나누고 후에 merge한다.

Point 3. 나뉜 Observable에 ObservableTransformer를 사용하여 원하는대로 오퍼레이터 파이프라인을 선언한다.

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;
import io.reactivex.observables.ConnectableObservable;


public class Main {
	public enum Type {
    	A, B, C
    }
    private static ObservableTransformer(Type, Type) typeAPipeline = new ObjservableTransformer<Type, Type>() {
    	@Override
        public ObservableSource<Type> apply(Observable<Type> observable) {
        	return observable
				.filter(new Predicate<Type>() {
        			@Override
            		public boolean test(Type arg0) throws Exception {
            		return arg0 == Type.A;
            		}
        		})            
            	.doOnNext(new Consumer<Type>(){
            		@Override
            		public void accept(Type arg0) throws Exception {
                		System.out.println("This is a pipeline only for Type A");
                	}
            	});
        }
    };
    private static ObservableTransformer(Type, Type) typeBPipeline = new ObjservableTransformer<Type, Type>() {
    	@Override
        public ObservableSource<Type> apply(Observable<Type> observable) {
        	return observable
				.filter(new Predicate<Type>() {
        			@Override
            		public boolean test(Type arg0) throws Exception {
            		return arg0 == Type.B;
            		}
        		})            
            	.doOnNext(new Consumer<Type>(){
            		@Override
            		public void accept(Type arg0) throws Exception {
                		System.out.println("This is a pipeline only for Type B");
                	}
            	});
        }
    };
    private static ObservableTransformer(Type, Type) typeCPipeline = new ObjservableTransformer<Type, Type>() {
    	@Override
        public ObservableSource<Type> apply(Observable<Type> observable) {
        	return observable
				.filter(new Predicate<Type>() {
        			@Override
            		public boolean test(Type arg0) throws Exception {
            		return arg0 == Type.C;
            		}
        		})            
            	.doOnNext(new Consumer<Type>(){
            		@Override
            		public void accept(Type arg0) throws Exception {
                		System.out.println("This is a pipeline only for Type C");
                	}
            	});
        }
    };
    public static void main(String[] args) throws Exception {
    	final Type types = {Type.A, Type.A, Type.B, Type.C, Type.A, Type.B, Type.B, Type.C};
        ConnectableObservable<Type> source = Observable.fromArray(types).publish();
        Observable<Type> typeAObservable = source.compose(typeAPipeline);
        Observable<Type> typeBObservable = source.compose(typeBPipeline);
        Observable<Type> typeCObservable = source.compose(typeCPipeline);
        Observable.merge(typeAObservable, typeBObservable, typeCObservable)
        			.subscribe();
        source.connect();
    }

}

※ 동일한 오퍼레이션을 두 개 이상 선언하면 doOn* 은 선언 순서대로, doAfter* 은 선언 역순으로 실행된다.

doOnComplete

마지막 요소가 방출될 때, onCompelete 바로 전에 실행됨. onComplete와 마찬가지로, 에러가 발생하면 실행되지 않음.

doOnDispose

구독을 해지했을 때 실행됨.

doOnNext

요소를 방출할 때, onNext 바로 전에 실행됨.

doOnError

오퍼레이션 중 에러가 발생했을 경우, onError 바로 전에 실행됨.

doOnEach

doOnComplete, doOnNext, doOnError의 통합 버전. Notification<?>을 아규먼트로 받아 isOnComplete, isOnNext, isOnError로 각각에 맞는 처리를 할 수 있음.

doOnSubscribe

구독 시, onSubscribe 바로 전에 실행됨.

doOnLifecyle

doOnSubscribe, doOnDispose의 통합 버전.

doOnTerminate

Observable이 종료될 때, onComplete 또는 onError 바로 전에 실행됨. doOnComplete 또는 doOnError가 선언되어 있다면 선언한 순서대로 실행됨. (업스트림 -> 다운스트림)

doAfterNext

요소를 방출할 때, onNext 바로 후에 실행됨.

doAfterTerminate

Observable이 종료될 때, onComplete 또는 onError 후에 실행됨. (cancel, dispose시에는 실행되지 않음) doFinnally가 선언되어 있다면 그 후에 실행됨.

doFinally

Observable이 종료될 때 실행됨. doAfterTerminate가 Observable의 완료 또는 에러에 대해서만 실행되지만, doFinally는 cancel, dispose시에도 실행된다. doAfterTerminate가 선언되어 있다면 doFinally가 먼저 실행된다.

실행 순서

doOnSubscribe -> doOnNext -> onNext -> doAfterNext -> 선언 순서에 따라 doOnComplete / doOnError 또는 doOnTermitate -> doFinally -> doAfterTerminate

subscribeOn, observeOn 이 두 가지의 차이에 대해 비교해 놓은 글이 상당히 많다. 

그냥 아래 다섯 가지만 기억하자.

1. observeOn 은 다운스트림에만 영향을 준다.

2. subscribeOn 은 업스트립, 다운스트림 양쪽에 영향을 준다.

3. 한 Observable에 대한 두 번 이상의 subscribeOn 호출은 첫 번째 호출만 유효하다.

4. 한 Observable에 대한 observeOn 호출은 몇 번이고 유효하다.

5. observeOn으로 변경된 쓰레드는 subscribeOn으로 변경할 수 없다.

출처:

https://proandroiddev.com/rx-java-subscribeon-and-observeon-a7d95041ce96

JUnit 5 는 JDK 8+ 을 요구한다. 

때문에 org.junit.jupiter.api.Assertions.. 를 사용하면 오류가 발생한다. (내부에서 Function 시리즈를 사용함.)

JUnit api 는 JUnit 4 이하의 org.junit.Assert... 를 사용해야 한다.

그리고 JUnit 5 에서, 이전 버전과의 런타임 호환성을 위한 vintage 를 의존성에 추가한다:

	testImplementation 'org.junit.jupiter:junit-jupiter-api:5.5.0'
	testImplementation 'org.junit.jupiter:junit-jupiter-api:5.5.0'
	testImplementation "junit:junit:4.12"
	testRuntimeOnly 'org.junit.vintage:junit-vintage-engine:5.5.0'
	testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.5.0'

5.6.0 이상을 사용하니 아래와 같은 오류가 발생했다. 고쳐서 사용할 수 있는 방법이 있을 수 있으나, 일단 5.5.0 을 사용하기로 한다.

* What went wrong: 
Execution failed for task ':compileTestJava'. 
> Could not resolve all files for configuration ':testCompileClasspath'. 
   > Could not resolve org.junit.jupiter:junit-jupiter-api:5.6.0. 
     Required by: 
         project : 
      > Cannot choose between the following variants of org.junit.jupiter:junit-jupiter-api:5.6.0: 
          - javadocElements 
          - sourcesElements 
        All of them match the consumer attributes: 
          - Variant 'javadocElements' capability org.junit.jupiter:junit-jupiter-api:5.6.0: 
              - Unmatched attributes: 
                  - Found org.gradle.category 'documentation' but wasn't required. 

Thymeleaf 를 사용하면 meta, link 태그 등의 닫기 태그가 없어 SAX 파싱 예외가 발생하는 경우가 있다. 일반적인 html에서는 저런 태그를 닫아주지 않아도 이런 경우가 없는데, 정확히 어느 버전까지인지는 모르겠지만, 예전 버전의 Thymeleaf lib 사용 시 xhtml의 엄격한 룰이 적용되어 이와 같은 현상이 발생한다고 한다.

이를 피하기 위해서는,

1. application.properties 에 다음 프로퍼티를 설정한다:

spring.thymeleaf.mode=LEGACYHTML5

2. nekohtml 의존성을 추가한다:

net.sourceforge.nekohtml:nekohtml

ps: nekohtml은 버전 1.9.15 이상을 사용하라는 권고가 있음.