너무 유용해서 정리 중

관련 책 : www.kyobobook.co.kr/product/detailViewKor.laf?ejkGb=KOR&mallGb=KOR&barcode=9788968486814

1. Redis의 이해

• 문서 : redis.io/
• 소스 : github.com/redis/redis
• 커맨드 : redis.io/commands

Redis의 주요 특징

• key-value 스토어 : 단순 스트링에 대한 Key/Value 구조를 지원한다.
• 컬렉션 지원 : List, Set, Sorted Set, Hash 등의 자료구조를 지원함
• Pub/Sub 지원 : Publish/Subscribe 모델을 지원한다 (서버간의 통지에 유용함)
• 디스크 저장 (Persistent Layer) :
  • 현재 메모리 상태의 스냅샷을 남기는 기능 'RDB' / 지금까지 실행된 업데이트 관련 명령어의 집합 'AOF'
  • RDB : 메모리 내용을 저장하는 기능 외에는 아무것도 지원하지 않는다. (데이터베이스가 아니다)
  • AOF (Append Only File) : set / del 등의 업데이트 관련 명령어를 그대로 기록함
• 복제 (replication) : 다른 노드에서 해당 내용을 복제할 수 잇는 마스터/슬레이브 구조를 지원
• 빠른 속도 : 이상의 기능을 지원하면서도 초당 100,000 QPS (Queries Per Second) 수준의 높은 성능을 자랑

Redis와 Memcached 비교

Memcached에 저장소의 개념이 추가된 것이 Redis 
저장소의 개념 == [RDB, AOF, Replication]

응답속도의 균일 성부분이 문제가 되는 이유 : 두개의 메모리 할당 구조가 다르기 때문
- Redis : jemalloc - free > 메모리 프래그멘테이션(fregmentation) 할당 비용 때문에 응답 속도가 느려진다.
- Memcached : slab 할당자 

2. Redis 운영과 관리

핵심 1 : Redis는 싱글 스레드다

싱글 스레드이기때문에 시간이 오래 걸리는 Redis 명령을 호출하면, 명령을 처리하는 동안에는 Redis가 다른 클라이언트의 요청을 처리할 수 없다.

1. 서버에서는 keys 명령을 사용하지 말자

keys 명령 : 원하는 패턴에 매칭되는 키들을 가져오는 명령어이다.
그러나 이걸 사용하면 장애로 이어질 가능성이 높다. 데이터 양이 늘어날 수록 해당 명령의 속도가 느려진다. 실제로 Redis 매뉴얼에 쓰지 말라고 되어있다.

redis.io/commands/keys

Warning
: consider KEYS as a command that should only be used in production environments with extreme care. It may ruin performance when it is executed against large databases. This command is intended for debugging and special operations, such as changing your keyspace layout. Don't use KEYS in your regular application code. If you're looking for a way to find keys in a subset of your keyspace, consider using  SCAN or SETS.

디버깅이나 스페셜한 명령어(keyspace layout 변경하기)로 의도된 것이라 보통의 어플리케이션 코드에서는 사용하면 안된다. 웬만하면 SCAN이나 SETS 명령어를 사용하는걸 고려해보아라.

모든 Key를 대상으로하는 명령어라서 그렇다.

2. flushall/flushdb 명령을 주의하자

db : 가상의 공간을 분리할 수 있는 개념   
- flushdb: db하나의 내용을 통째로 지우는 것
- flushall : 모든 db의 내용을 모두 지우는 것

flushall 명령은 전체 데이터를 다 지우며 keys명령처럼 많은 시간이 필요하다 (memcached는 순식간에 모든 데이터가 지워진다) - 동작방식이 다르다

아이템 개수에 비례해서 시간이 걸린다
실제 데이터를 일일히 삭제하는 로직으로 구현이 되어있다 : 지우는 속도가 O(n)

Memecached의 flush_all이 빠른 이유

Memcached에서는 실제로 데이터를 삭제하지 않는다.
해당 명령어가 실행된 시간만 기록하고, 이보다 이전에 저장된 key는 get명령을 통해서 접근할 때 없다고하면서 실제로는 지운다. (oldest_live 플래그만 세팅 : 이보다 먼저 생성된 key는 없다)

핵심2: Redis Persistent

Reids Persistent : Redis 데이터를 디스크로 저장할 수 있다 > 디스크에 저장된 데이터 기반으로 복구가 가능하다

1. RDB

현재 메모리에 대한 덤프를 생성하는 기능.

fork를 이용해서 자식 프로세스를 생성한다. > 현재 메모리 상태가 복제된 자식프로세스를 기반으로 데이터를 저장한다 (지속적인 서비스가 가능하다) - 근데 이렇게 하면 최악의 상황에 메모리가 2배가 필요할 것 같다.

- SAVE : 모든 작업을 멈추고 현재 메모리 상태에 대한 RDB파일 생성 : 싱글 스레드 이므로 아무 작업도 수행할 수 없음
- BGSAVE : 백그라운드 SAVE : fork 작업을 통해 자식 프로세스에서 파일을 저장한다.

#redis.conf
dbfilename dump.rdb

2. AOF

AOF : Append Only File
현재 수행해야 할 명령을 미리 저장해두고, 장애가 발생하면 AOF 기반으로 복구

1. 클라이언트가 Redis에 업데이트 관련 명령 요청
2. Redis가 해당 명령을 AOF에 저장
3. 파일 쓰기가 완료되면 실제로 해당 명령을 실행해서 메모리의 내용을 변경함.

AOF는 기본적으로 사용안함으로 되어있어 변경해주어야한다.

#redis.conf
appendonly yes # default no
appendfilename appendonly.aof # 파일이름 설정
appendfsync everysec # 디스크 동기화를 얼마나 자주할 것인가 (always, everysec, no)

AOF와 RDB의 우선순위 : 최신 데이터를 더 많이 가진 파일 - AOF 
RDB는 스냅샷인 반면, AOF는 매 작업마다 디스크에 기록을 남겨서 모든 데이터가 남아있다.

참고 : 레디스 프로토콜

*키워드개수\r\n
[키워드개수만큼 반복]
$키워드크기\r\n
키워드\r\n

3. Redis가 메모리를 두배로 사용하는 문제

장애 원인 : RDB저장시 fork를 사용하기 때문에.

COW (Copy on Write) : 실제로 변경이 발생한 부분만 차후에 복사한다. (그러나 redis는 부분 write가 많다)

참고 : redisgate.kr/redis/configuration/copy-on-write.php

참고 : 대용량 메모리와 Redis

Redis는 싱글 스레드임 : 멀티 코어를 활용하기 위해 여러 개의 Redis 서버를 한 서버에 띄우는 것이 성능 면에서 좋다.

Core4개 + 메모리 32GB 장비 : 프로세스별로 6G 할당하기
- 평상시 : 6 + 6 + 6 + 6 = 24GB
- fork 복제시 : 6 + 6 + 6 + 6 + (6 = 자식) = 32GB

4. Redis장애 : Read 성공 Write 실패

Redis 기본 설정 : RDB저장 실패 => 장비 이상으로 판단 => Write 명령처리하지 않음 
lastbgsave_status = REDIS_ERR : Write 관련 요청 모두 무시

Heartbeat체크는 읽기 관련 명령을 이용해 검사하기 때문에 문제 발생 가능.

RDB 생성 실패의 경우
- RDB를 저장할 수 있을 정도의 디스크 여유 공간이 없는 경우
- 실제 디스크가 고장 난 경우
- 메모리 부족으로 인해 자식 프로세스를 생성하지 못한 경우
- 누군가 강제적으로 자식 프로세스를 종료시킨 경우

해결 방법
1) 해당 상황이 맞는지 확인하기 : set 명령어 사용
2) info 명령어 사용 : rdb_last_bg_save_status:ok 인지 확인
3) 정책 결정 : config set stop-writes-on-bgsave-error no

3. Redis 복제

1. Redis 복제 모델

redis는 마스터 / 슬레이브 형태의 복제 모델 제공
-> 이때 슬레이브가 다른 장비의 마스터로도 동작할 수 있게도 할 수 있음 (M-S-S)

1. 명령어 사용 : slaveof <ip> <port>
2. redis.conf 수정 : slaveof <master ip> <master port>

redis-cli로 접속 후 info 명령어 혹은 info replication 명령어를 사용한다.

2. Redis 복제 과정

1. slave에서 slaveof 명령어를 이용해 마스터 서버를 설정한다.
2. 마스터 서버가 설정되면 replicationCron에서 현재 상태에 따라 connectWithMaster를 호출한다.
3. 마스터는 복제를 위해 RDB를 생성한 후 슬레이브에 전송한다. 
4. 슬레이브는 RDB를 로드하고, 나머지 차이에 대한 명령을 마스터에서 전달받아 복제를 완료한다.

RDB 기준 복제 순서 (full synchronization)
1. 마스터는 자식 프로세스를 시작해 백그라운드로 RDB파일에 데이터를 저장합니다.
2. 데이터를 저장하는 동안 마스터에 새로 들어온 명령들은 처리 후 복제버퍼에 저장됩니다.
3. RDB 파일 저장이 완료되면, 마스터는 파일을 복제서버에게 전송합니다.
4. 복제서버는 파일을 받아 디스크에 저장하고, 메모리로 로드합니다.
5. 마스터는 복제버퍼에 저장된 명령을 복제서버에게 전송합니다.

이때 복제중 끊어지게 되면 backlog-buffer에 데이터가 저장된다. 이후 다시 연결되었을때 이 buffer에서부터 동기화가 이루어진다.
만약 buffer가 넘쳤을때는 full synchronization을 진행한다

3. Redis 복제 사용시 주의 사항

주의점 1. slaveof no one을 기억하자

슬레이브는 마스터의 상태를 지속적으로 감시하면서 바뀌는 내용을 계속 전달받는다. 
- 연결 상태 이상이 생기는 경우 : 재 연결이 되면 rsync
- 마스터에 장애가 발생하여 마스터에 데이터가 없는경우 : 슬레이브의 모든 내용이 사라진다.

슬레이브 복제를 진행할 때 emptyDb() 호출로, 현재 자신(슬레이브)의 데이터를 모두 삭제하고 마스터와의 싱크를 맞추려하기 때문이다.

slaveof no one : 더이상 슬레이브로 동작하지 않도록 설정하는 것

주의점 2. 복제 시에 무조건 RDB를 백그라운드로 생성한다는 것을 주의하자

RDB를 사용한다 == 메모리를 두배로 사용할 가능성이 있다. > 설정을 꺼둠
그치만 복제를 사용한다는 것 자체가 fork를 사용해서 RDB를 생성한다는 것 : 하나의 프로세스가 너무 많은 메모리를 사용하지 않도록 나누는 과정이 필요하다.

4. Redis 복제를 이용한 실시간 마이그레이션

1. 데이터 이전을 위한 새로운 redis 인스턴스 실행
2. 새로운 Redis 인스턴스를 기존 마스터의 슬레이브로 설정 (slaveof) - 복제 완료
3. 새로운 장비의 slave-read-only 설정을 끈다.
4. 클라이언트들이 새로운 redis 인스턴스를 마스터로 인식하도록 설정을 바꾼다.
5. slave no one 명령어 : 기존 마스터와의 연결 종료 (완전한 마이그레이션 이후)
6. 기존 장비 제거

4. Redis HA와 Sentinel

1. Redis HA와 Sentinel 구성

redis master에 장애가 발생하면 sentinel은 슬레이브 중 한대를 선택해서 마스터로 승격시킨다 (내부적으로 투표과정을 거친다)
이때 sentinel이 client에 pub/sub으로 master 변경을 통지한다. 
- client(sub) -> sentinel(pub)

2. Sentinel이 장애를 판별하는 방법

기본 : PING 명령어의 응답을 활용해서 판단함. (PONG이 안와도 바로 장애로 판단하지는 않는다)
- SDOWN : Subjectively Down : 해당 서버의 장애를 주관적으로 판단 
- ODOWN : Objectively Down : 해당 서버의 장애를 객관적으로 판단 -> 진짜 장애 : Failover 진행한다.

sentinel이 여러대 있을 때 Quorum(쿼럼) 값 이상의 센티널에서 SDOWN으로 판단해야 ODOWN이 된다.
- 주로 Sentinel 장비 수를 홀수로
- 쿼럼 값을 해당 장비수의 과반으로 설정하는 것이 좋다
   > 대부분의 설명에서 Sentinel을 3대를 띄우고 쿼럼 값을 2정도로 주는 것으로 얘기를 하긴 한다.

SDOWN 판별 : 해당 서버의 last_avail_time과 현재 시간의 차이가 설정된 down_after_period 값보다 클 때
ODOWN 판별 : SDOWN일 경우에만 쿼럼 값을 체크하며, 이 이상일때 ODOWN

3. Sentinel이 마스터로 승격할 슬레이브를 선택하는 방법

sentinelSelectSlave 함수로 결정

1. SDOWN, ODOWN, DISCONNECT 된 상태의 슬레이브는 제외
2. last_avail_time이 info_validity_time 보다 작으면 제외
3. info_refresh 값이 info_validity_time 보다 작으면 제외
4. master_link_down_time 이 max_master_down_time 보다 크면 제외
5. 남은 후보들 중에서 slave_priority가 높은 슬레이브가 우선적으로 선택

slave_priorirty 값은 redis.conf 파일을 조정하여 선출시의 가중치에 영향을 줄 수 있다. 
- 0으로 설정하면 해당 슬레이브는 절대로 마스터 승격이 안된다.

4. Sentinel 설정과 사용

# sentinel monitor <클러스터 명> <마스터 IP> <마스터 Port> <쿼럼값>
sentinel monitor resque 127.0.0.1 2001 2 

# 다운으로 인식하는 시간
# sentinel down-after-milliseconds <클러스터 명> <시간 miliseconds>
sentinel down-after-milliseconds resque 3000 

# sentinel failover-timeout <클러스터 명> <시간 miliseconds>
sentinel failover-timeout resque 900000

# sentinel can-failover <클러스터 명>
sentinel can-failover resque yes # failover 여부 설정

# 마스터 승격후 몇개의 슬레이브가 싱크를 해야 클라이언트에 알려줄 것인지
# sentinel parallel-syncs <클러스터 명> <sync할 slave 숫자>
sentinel parallel-syncs resque 1 

psubscribe 명령을 이용해 통지를 받는다. (pub/sub)

+switch-master 감지 
<클러스터 명> <이전 마스터 IP> <이전 마스터 Port> <새 마스터 IP> <새 마스터 Port>

5. Redis 모니터링

python 스크립트 : info 명령어를 활용한 스크립트. (python-redis 모듈)

Percona Cacti 플러그인 : www.percona.com/doc/percona-monitoring-plugins/LATEST/cacti/redis-templates.html

Redis-stat : github.com/junegunn/redis-stat

redmon : github.com/steelThread/redmon

추가 참고하기 좋은 사이트 - redisgate.kr/redisgate/ent/ent_intro.php

소스코드

https://github.com/mjung1798/Jyami-Java-Lab/tree/master/java8-in-action

리액티브 프로그래밍에서는 다양한 시스템과 소스에서 들어오는 데이터 항목 스트림을 비동기적으로 처리하고 합쳐서 문제를 해결한다
- 빅데이터 : 빅데이터는 페타 바이트 단위로 구성되며 매일 증가한다
- 다양한 환경 : 모바일 디바이스 부터 수천개의 멀티 코어 프로세서로 실행되는 클라우드 기반 클러스터까지 다양한 환경에 배포된다.
- 사용패턴 : 사용자는 1년 내내 항상 서비스를 이용할 수 있으며 밀리초 단위의 응답 시간을 기대한다.

리액티브 프로그래밍 패러다임에 맞게 설계된 애플리케이션은 높은 응답성을 제공한다.
전체의 리액티브 시스템을 구성하는 여러 컴포넌트를 조절에서도 가용성을 제공한다.

1. 리액티브 매니페스토

www.reactivemanifesto.org/

리액티브 매니페스토에서는 리액티브 애플리케이션과 시스템 개발의 핵심 원칙을 공식적으로 정의한다. 

• 반응성(responsive) : 시스템은 가능한한 적정시간 안에 반응한다. 반응성이 뒷받침 되어야 사용성을 높일 수 있다.
• 탄력성(elastic) : 다양한 작업 부하에도 시스템 반응성이 유지된다. 입력속도가 바뀐다면 이들 입력 관련 서비스에 할당된 자원을 늘리거나 줄임으로 반응할 수 있다.
• 메시지 주도(message driven) : 컴포넌트 간의 약산 결합, 고립, 위치, 투명성이 유지되도록 시스템은 비동기 메시지 전달에 의존한다.
• 회복성(resilient) : 장애 시에도 시스템의 반응성은 유지된다.

1. 애플리케이션 수준의 리액티브

애플리케이션 수준에서의 주요기능은 비동기로 작업을 수행할 수 있다는 점이다.
비동기로 처리하는 것이 최신 멀티코어 CPU 사용률을 극대화(내부적으로 경쟁하는 CPU의 스레드 사용률) 할 수 있는 방법이다

이를 위해 스레드를 퓨처, 액터, 일련의 콜백을 발생시키는 이벤트 루프 등과 공유하고 처리할 이벤트를 변환하고 관리한다.

개발자 입장에서는 동기 블록, 경쟁 조건, 데드락 같은 저 수준의 멀티스레드 문제를 직접 처리할 필요가 없어져 비즈니스 요구사항 구현에 더 집중이 가능하다.

RxJava, Akka같은 리액티브 프레임워크는 별도로 지정된 스레드 풀에서 블록 동작을 실행시킨다. > 메인 풀의 모든 스레드는 방해 받지 않고 실행된다. (CPU 관련 작업과 IO관련 작업의 분리)

2. 시스템 수준의 리액티브

리액티브 시스템  = 소프트웨어 아키텍처 : 여러 애플리케이션이 한 개의 일관적인, 회복할 수 있는 플랫폼을 구성할 수 이쎅 해줄 뿐 아니라 이들 애플리케이션 중 하나가 실패해도 전체 시스템이 계속 운영되게 해준다.

- 리액티브 애플리케이션 : 이벤트 주도
- 리액티브 시스템 : 메시지 주도

컴포넌트에서 발생한 장애를 고립시킴으로 문제가 주변의 다른 컴포넌트로 전파되면서 전체 시스템 장애로 이어지는 것을 막음으로 회복성을 제공한다(== 결함 허용 능력) : 고립과 비결합

모든 컴포넌트가 수신자의 위치에 상관없이 모든 서비스와 통신 ( 탄력성 = 위치 투명성)

2. 리액티브 스트림과 플로 API

리액티브 프로그래밍 = 리액티브 스트림을 사용하는 프로그래밍
리액티브 스트림 : 잠재적으로 무한의 비동기 데이터를 순서대로 처리하고, 블록하지 않는 역압력을 전제해 처리하는 표준 기술

역압력 : pub-sub 프로토콜에서 publisher가 발행하는 속도 > subscriber가 소비하는 속도 일 경우에 문제가 발생하지 않도록 보장하는 장치 + 기존 데이터 처리에 얼마나 시간이 걸리는지 업스트림 발행자에게 알릴 수 있어야함

데이터 수신자가 스레드를 블록하지 않고도 데이터 수신자가 처리할 수 없을 만큼의 데이터를 받는 일을 방지한다.

리액티브 스트림이 구현이 제공해야 하는 최소 기능 집합 (네개 인터페이스)

• java9의 새로운 java.util.concurrent.Flow 클래스
• Akka 스트림, 리액터(reactor), RxJava, Vert.x 등 많은 서드 파티 라이브러리

2-1. Flow 클래스

Publisher가 항목을 발행하면 Subscriber가 한 개씩 또는 한 번에 여러 항목을 소비하는데 Subscription이 이 과정을 관리하도록 여러 정적 메서드를 제공한다.

public final class Flow {
    static final int DEFAULT_BUFFER_SIZE = 256;

    private Flow() {
    }

    public static int defaultBufferSize() {
        return 256;
    }

    public interface Processor<T, R> extends Flow.Subscriber<T>, Flow.Publisher<R> {
    }

    public interface Subscription {
        void request(long var1);

        void cancel();
    }

    public interface Subscriber<T> {
        void onSubscribe(Flow.Subscription var1);

        void onNext(T var1);

        void onError(Throwable var1);

        void onComplete();
    }

    @FunctionalInterface
    public interface Publisher<T> {
        void subscribe(Flow.Subscriber<? super T> var1);
    }
}

• Publisher : 많은 이벤트 제공이 가능하지만 Subscriber의 요구사하에 따라 역압력 기법에 의해 이벤트 제공 속도가 제한
• Subscriber : Publisher가 발행한 이벤트의 리스너로 자신을 등록할 수 있음.

onSubscribe onNext* (onError| onComplete)?

- onSubscribe 메서드는 항상 처음 호출
- onNext 메서드는 여러번 호출 가능
- 이벤트 스트림이 영원히 지속되거나 onComplete, onError 호출 가능

규칙

• Publisher는 Subscription의 request 메서드에 정의된 개수 이하의 요소만 Subscriber에 전달한다.
• Publisher는 동작이 성공적으로 끝나면 onComplete, 문제가 발생하면 onError를 호출해 Subscription을 종료한다.
• Subscriber는 요소를 받아 처리할 수 있음을 Publisher에게 알려야 한다. (역압력 행사)
• onComplete나 onError를 처리하는 상황에서 Subscriber는 Publisher나 Subscription의 어떤 메서드도 호출 할수도없고 Subscription이 취소되었다 가정한다.
• Subscriber는 Subscription.reqeust() 호출 없이도 언제나 종료 시그널을 받을 준비가 되어있어야한다.
• Subscriber는 Subscription.canel() 이 호출된 이후에도 한 개 이상의 onNext를 받을 준비가 되어있어야 한다.

이 명세에 맞춰 직접 구현한 기능은 Reative Streams TCK라는 툴로 검증할 수 있다. 구현이 까다로움..

Processor 인터페이스 ; Publisher, Subscriber 상속
리액티브 스트림에서 처리하는 이벤트의 변환 관계를 나타낸다.(?)

Flow 클래스의 인터페이스는 직접 구현하도록 의도된게 아니다. (실제로 자바9에서 구현 클래스를 제공하지않는다.)
라이브러리가 준수해야할 규칙과 리액티브 애플리케이션이 서로 협동 소통할 수 있는 공용어 제시가 목표였기 때문이다.
Akka, RxJava 등의 리액티브 라이브러리에서 이미 구현체부터 만들어둔 상태였기 때문에.

2-2. 간단한 리액티브 애플리케이션

@Getter
public class TempInfo {
    public static final Random random = new Random();

    private final String town;
    private final int temp;

    public TempInfo(String town, int temp) {
        this.town = town;
        this.temp = temp;
    }

    public static TempInfo fetch(String town){
        if(random.nextInt(10) == 0)
            throw new RuntimeException("Error!");
        return new TempInfo(town, random.nextInt(100));
    }

    @Override
    public String toString() {
        return "TempInfo{" +
                "town='" + town + '\'' +
                ", temp=" + temp +
                '}';
    }
}

public class TempSubscriber implements Flow.Subscriber<TempInfo> {

    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1); // 구독 저장하고 첫번째 요청 전달 (1개의 요청을 받을 준비가 되어있다.)
    }

    @Override
    public void onNext(TempInfo tempInfo) {
        System.out.println(tempInfo); // 수신한 온도 출력하고 다음 정보 요청
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.err.println(throwable.getMessage()); // 에러 메세지 출력
    }

    @Override
    public void onComplete() {
        System.out.println("Done!");
    }
}

public class TempSubscription implements Flow.Subscription {

    private final Flow.Subscriber<? super TempInfo> subscriber;
    private final String town;

    public TempSubscription(Flow.Subscriber<? super TempInfo> subscriber, String town) {
        this.subscriber = subscriber;
        this.town = town;
    }

    @Override
    public void request(long n) {
        for (long i = 0L; i < n; i++) {
            try {
                subscriber.onNext(TempInfo.fetch(town)); // 현재 온도를 Subscriber로 전달
            }catch (Exception e){
                subscriber.onError(e); // 온도 가져오기를 실패하면 Subscriber로 에러 전달
                break;
            }
        }
    }

    @Override
    public void cancel() {
        subscriber.onComplete(); // 구독이 취소되면 완료 신호를 Subscriber에 전달
    }
}

public class Main {
    public static void main(String[] args){
        getTemperatures("Seoul").subscribe(new TempSubscriber()); // 서울에 Publisher를 만들고 TempSubscriber를 구독
    }

    private static Flow.Publisher<TempInfo> getTemperatures(String town){ // Publisher.subscribe() 메서드 구현
        return subscriber -> subscriber.onSubscribe(new TempSubscription(subscriber, town));
        // 구독한 Subscriber에게 TempSubscription을 전송하는 Publisher를 반환
    }
}

1. Publisher는 functional 인터페이스이며, 인자로 subscriber를 받는다.
2. Publisher는 인자로 받은 subscriber의 onSubscribe 메서드를 호출한다.
3. onSubscribe 메서드 호출로 subscriber는 subscription 객체를 인스턴스 변수로 갖고있게 된다.
4. 이 subscription 객체는 subscriber를 인스턴스 변수로 갖고있게 된다.
5. onSubscribe 메서드 호출과 함께, subscription.request(1) 이 불리게 되면서, subscriber가 1개의 요청을 받을 준비가 되었음을 subscription이 알게된다.
6. subscription.request(1)로 subscription 내의 인스턴스 변수인 subscriber의 onNext 혹은 onError 가 불리게된다.
7. onNext의 경우에 마찬가지로 subscription.request(1) 이 불리면서 subscriber가 1개의 요청을 받을 준비가 되었음을 subscription이 알게된다.
8. onError의 경우에는 더이상 subscription에 request 가 일어나지 않아 종료된다.

즉. Publisher의 subscribe로 인해 계속 이벤트가 방출되게 되고, 이 이벤트를 Subscription의 request 요청이 있어야지만 Subscriber 가 이벤트를 소비하게된다.

이때 에러를 발생시키는 코드를 없애게되면 재귀호출로인한 stackoverflow가 발생한다. : Executor를 TempSubscription으로 추가한 다음 다른 스레드에서 TempSubscriber로 전달한다. 

private static final ExecutorService executor = Executors.newSingleThreadExecutor();

    @Override
    public void request(long n) {
        executor.submit(() -> {
            for (long i = 0L; i < n; i++) {
                try {
                    subscriber.onNext(TempInfo.fetch(town)); // 현재 온도를 Subscriber로 전달
                }catch (Exception e){
                    subscriber.onError(e); // 온도 가져오기를 실패하면 Subscriber로 에러 전달
                    break;
                }
            }
        });
    }

3. Processor로 데이터 변환하기

Processor는 Subscriber이며 Publisher이다.

public class TempProcessor implements Flow.Processor<TempInfo, TempInfo> {

    private Flow.Subscriber<? super TempInfo> subscriber;

    @Override
    public void subscribe(Flow.Subscriber subscriber) {
        this.subscriber = subscriber;
    }

    @Override
    public void onNext(TempInfo tempInfo) {
        int fTemp = (((tempInfo.getTemp() - 32) * 5) / 9); // 섭씨로 변환한 다음 TempInfo 다시 전송
        subscriber.onNext(new TempInfo(tempInfo.getTown(), fTemp));
    }

    // 다른 모든 신호는 업스트림 구독자에게 전달.
    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        subscriber.onSubscribe(subscription);
    }

    @Override
    public void onError(Throwable throwable) {
        subscriber.onError(throwable);
    }

    @Override
    public void onComplete() {
        subscriber.onComplete();
    }
}

    @Test
    @DisplayName("processor를 이용한 C=>F 온도 변환")
    void name() {
        getCelsiusTemperatures("Seoul").subscribe(new TempSubscriber());
    }

    public static Flow.Publisher<TempInfo> getCelsiusTemperatures(String town){
        return subscriber -> {
            TempProcessor processor = new TempProcessor();
            processor.subscribe(subscriber);
            processor.onSubscribe(new TempSubscription(processor,town));
        };
    }

Subscriber 인터페이스를 구현하는 다른 모든 메서드는 단순히 수신한 모든 신호를 업스트림 Subscriber로 전달하며 Publisher의 subscribe 메서드는 업스트림 Subscriber를 Processor로 등록하는 동작을 수행한다.

4. 자바에서 플로 API의 구현을 제공하지 않는 이유

API를 만들 당시 Akka, RxJava 등 다양한 리액티브 스트림의 자바 코드 라이브러리가 이미 존재했기 때문이다. 둘다 같은 pub-sub 패턴을 기반한 리액티브 프로그래밍을 구현했지만 이들 라이브러리는 독립적으로 구현이 되었고 각기다른 이름 규칙과 API를 사용함

그래서 플로 API에서 리액티브 개념의 구현을 위한 표준화 작업을 하게된 것이다.

리액티브 스트림의 구현은 대부분 기존 구현을 따라가자.

3. 리액티브 라이브러리 RxJava 사용하기

RxJava는 자바로 리액티브 애플리케이션을 구현하는 데 사용하는 라이브러리이다.

github.com/ReactiveX/RxJava

reactivex.io/

RxJava 에서 제공하는 Flow.Publisher 구현 클래스

1. Flowable

public abstract class Flowable<@NonNull T> implements Publisher<T> {}

역압력 기능이 있다. : 너무 빠른 속도로 데이터를 발행해 Subscriber가 이를 감당할 수 없는 상황에 이르는걸 방지하는 기능

2. Observable

public abstract class Observable<@NonNull T> implements ObservableSource<T> {}

역압력 기능을 제공하지 않던 기존버전의 RxJava에서 제공하던 클래스

단순한 프로그램, 마우스 움직임 같은 사용자 인터페이스에 적합. (GUI이벤트나 자주 발생하지 않는 종류의 이벤트에 역압력을 적용하지 말자.)

모든 Publisher는 Subscription의 request(Long.MAX_VALUE); 메서드를 이용해 역압력 기능을 끌 수 있다.

3-1. Observable 만들고 사용하기

Observable과 Flowable 클래스는 다양한 종류의 리액티브 스트림을 편리하게 만들도록 여러 팩토리 메서드를 제공한다
이들은 모두 Publisher를 구현하므로 팩토리 메서드는 리액티브 스트림을 만드는 것이다.

    @Test
    @DisplayName("Observable just 메서드 : Observable의 Subscriber는" +
            " onNext(first) onNext(second) onComplete() 순서로 메서드를 받는다.")
    void observableJust() {
        Observable<String> just = Observable.just("first", "second");
    }

reactivex.io/documentation/operators/just.html

    @Test
    @DisplayName("Observable interval 메서드 : 지정된 속도로 이벤트를 방출하는 상황에서 사용한다." +
            "0에서 시작해 1초 간격으로 long 형식의 값을 무한으로 증가시키며 값을 방출한다.")
    void observableInterval() {
        Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);
    }

reactivex.io/documentation/operators/interval.html

RxJava에서의 Flow.Subscriber 역할

public interface Observer<@NonNull T> {

    void onSubscribe(@NonNull Disposable d);

    void onNext(@NonNull T t);

    void onError(@NonNull Throwable e);

    void onComplete();

}

Subscriber과 같은 메서드를 정의하며, onSubcribe 메서드가 Subscription 대신 Disposale 인수를 갖는다는 점만 다르다.
Observable은 역압력을 지원하지 않아 Subscription의 request가 필요하지 않다.

하지만 onNext가 좀더 유연하다 (더 많은 오버로드 기능 제공) : ex onNext만 구현하고 나머지는 구현하지 않는 기본동작의 Observer 제공가능.

    @Test
    @DisplayName("RxJava Observer: 많은 오버로드로 인해 유연하게 구현이 가능하다.")
    void observableIntervalWithObserver() {
        Observable<Long> onePerSec = Observable.interval(1, TimeUnit.SECONDS);
        onePerSec.subscribe(i -> System.out.println(TempInfo.fetch("Seoul")));
    }

1초에 한번씩 TempInfo의 temp 값을 출력하리라 예상한다.

그렇지만 Observable이 RxJava의 연산 스레드 풀인 데몬 스레드에서 실행되어, 위 코드에서는 실행하자마자 실행할 코드가 없어 종료되게 된다. 데몬스레드도 함께 종료 > blockingSubscribe를 사용해 결과를 볼 순 있다.

심화예제

    @Test
    @DisplayName("좀더 일반화를 위해 온도를 직접 출력하지 않고 " +
            "사용자에게 팩토리 메서드를 제공해 매초마다 온도를 방출하는 Observable을 만들어보자" +
            "이렇게 할 경우 observer는 그저 받은 데이터만 출력하는 기능을 하면된다." +
            "observable로 온도를 얻는 과정에서 에러와 완료처리 로직을 다 구현했기 때문에 ")
    void observableExample() {
        Observable<TempInfo> seoul = getTemperature("Seoul");
        seoul.blockingSubscribe(new TempObserver());
    }

    public static Observable<TempInfo> getTemperature(String town){
        return Observable.create(emitter ->
                Observable.interval(1, TimeUnit.SECONDS) // 매 초마다 무한으로 증가하는 long 반환 Observable
                        .subscribe(i -> {
                            if(!emitter.isDisposed()){ // 소비된 옵저버가 폐기되지 않았으면 어떤 작업을 수행(에러가 안났으면)
                                if(i >= 5){ // 5번 온도를 onNext 했으면 성공처리 후 종료
                                    emitter.onComplete();
                                }else {
                                    try {
                                        emitter.onNext(TempInfo.fetch(town)); // 온도를 observer로 보
                                    }catch (Exception e){
                                        emitter.onError(e);// 에러가 발생하면 Observer에게 알림
                                    }
                                }
                            }
                        })
        );
    }
    
    public class TempObserver implements Observer<TempInfo>{

        @Override
        public void onSubscribe(@NonNull Disposable d) {
        }

        @Override
        public void onNext(@NonNull TempInfo tempInfo) {
            System.out.println(tempInfo);
        }

        @Override
        public void onError(@NonNull Throwable e) {
            System.err.println("Got Problem " + e.getMessage());
        }

        @Override
        public void onComplete() {
            System.out.println("Done!");
        }
    }

여기서 사용된 emitter 변수는 그 실체가 ObservableEmitter 인터페이스인데, 그 인터페이스는 Emitter의 인터페이스를 상속한다.. : onSubscribe가 빠진 Observer와 같음.

public interface ObservableEmitter<@NonNull T> extends Emitter<T> {}

public interface Emitter<@NonNull T> {

    void onNext(@NonNull T value);

    void onError(@NonNull Throwable error);

    void onComplete();
}

여기서 Observable이 역압력을 지원하지 않아 전달된 요소를 처리한 다음 추가 요소를 요청하는 request()가 필요가 없다.

3-2. Observable을 변환하고 합치기

리액티브 라이브러리는 자바 9 플로 API에 비해 스트림을 합치고 만들고 거르는 등의 풍부한 메서드를 제공하는 것이 장점이다. 스트림을 다른 스트림의 입력으로 사용도 가능하며, 스트림의 매핑 함수로 요소를 변환하거나, 다양한 방법으로 합치는 등의 작업도 가능하다.

마블 다이어그램(marble diagram) : 수평선으로 표시된 리액티브 스트림에 임의의 순서로 구성된 요소가 기하학적 모양으로 나타난다.
- 특수기호: 에러나 완료신호를 나타낸다.
- 박스 : 연산이 요소를 어떻게 변화하거나 여러 스트림을 어떻게 합치는지를 보여줌

reactivex.io/documentation/operators

RxJava의 많은 operator를 설명하는 페이지에도 마블 다이어그램을 이용해 여러 연산의 스트림 동작을 설명한다.

merge - 두개 이상의 Observable이 방출한 이벤트를 하나로 합침
map - Observable이 발행하는 요소를 변환

    @Test
    @DisplayName("Observable의 연산자를 이용해 스트림을 유동적으로 처리할 수 있다." +
            "filter 를 이용해 섭씨온도 영하만 필터링하고 " +
            "map을 이용해 화씨를 섭씨로 변환한다")
    void getMinusCelsiusTemperature() {
        Observable<TempInfo> seoul = getTemperature("Seoul")
                .filter(temp -> temp.getTemp()<0)
                .map(temp -> new TempInfo(temp.getTown(), ((temp.getTemp() - 32) * 5) / 9));
        seoul.blockingSubscribe(new TempObserver());
    }

소스코드

https://github.com/mjung1798/Jyami-Java-Lab/tree/master/java8-in-action

5. 발행-구독 그리고 리액티브 프로그래밍

리액티브 프로그래밍은 Future같은 객체를 통해 여러 결과를 제공 (future는 한번만 실행해 결과를 제공)

자바 9에서 java.util.concurrent.Flow 인터페이스에 발행-구독 모델(pub-sub 이라 불리는 프로토콜)을 적용해 리액티브 프로그래밍을 제공한다.

• 구독자(subscriber)가 구독할 수 있는 발행자(publisher)
• 이 연결은 구독(subscription)이라 한다
• 이 연결을 이용해 메세지(또는 이벤트로 알려짐)을 전송한다.

C3=C1+C2 스프레드시트 셀을 만든다 하자. (C1, C2 값이 갱신되면 C3에도 새로운 값이 반영된다)
- c1, c2에 이벤트가 발생했을 때 c3을 구독하도록 한다. > Publisher<T> 필요
- Publisher<T>는 통신할 구독자(Subscriber)를 인수로 받는다

//
// Source code recreated from a .class file by IntelliJ IDEA
// (powered by Fernflower decompiler)
//

package java.util.concurrent;

public final class Flow {
    static final int DEFAULT_BUFFER_SIZE = 256;

    private Flow() {
    }

    public static int defaultBufferSize() {
        return 256;
    }

    public interface Processor<T, R> extends Flow.Subscriber<T>, Flow.Publisher<R> {
    }

    public interface Subscription {
        void request(long var1);

        void cancel();
    }

    public interface Subscriber<T> {
        void onSubscribe(Flow.Subscription var1);

        void onNext(T var1);

        void onError(Throwable var1);

        void onComplete();
    }

    @FunctionalInterface
    public interface Publisher<T> {
        void subscribe(Flow.Subscriber<? super T> var1);
    }
}

Cell은 Publisher인 동시에 Subscriber이다. (셀 이벤트에 구독을 할 수 있다 / 다른 셀의 이벤트에 반응한다)

public class SimpleCell implements Flow.Publisher<Integer>, Flow.Subscriber<Integer> {
    private int value =0;
    private String name;
    private List<Flow.Subscriber> subscribers = new ArrayList<>();

    public SimpleCell(String name) {
        this.name = name;
    }

    // Publisher 메서드
    @Override
    public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
        subscribers.add(subscriber);
    }

    //Subscriber 메서드
    @Override
    public void onNext(Integer newValue) {
        this.value = newValue; // 구독한 셀에서 새로운 값이 생겻을 때 값을 갱신한다.
        System.out.println(this.name + ":" + this.value);
        notifyAllSubscribers(); // 값이 갱신됌을 모든 구독자에게 알림
    }

    private void notifyAllSubscribers(){ // 새로운 값이 있음을 모든 구독자에게 알리는 메서드
        subscribers.forEach(subscriber -> subscriber.onNext(this.value));
    }
}

    @Test
    @DisplayName("구독관계 확인 : C1은 Publisher / C3은 Subscriber > publisher 값이 변하면 subscriber에 이벤트가 간다")
    void simpleCell() {
        SimpleCell c3 = new SimpleCell("C3");
        SimpleCell c2 = new SimpleCell("C2");
        SimpleCell c1 = new SimpleCell("C1");

        c1.subscribe(c3);

        c1.onNext(10);
        c2.onNext(20);

//        C1:10
//        C3:10
//        C2:20
    }

이 구조에서 (Subscription을 사용하지 않은 구조)에서는 publisher, subscriber의 동작을 연결하기 위해 참조객체로 List<Subsriber> subscribers을 두었다. 
c1은 publisher이면서(subscribe 명령어 수행) subscriber이다(onNext 수행) 

> 여기서는 역압력(backpressure) 내용이 나오지 않았음

기존의 옵저버 패턴에 비해 새로운 API 프로토콜이 더 강력해진 이유 ; onNext, onError, onComplete와 같은 메서드로 데이터 흐름에서 예외가 발생하거나 종료되었을 경우도 제어하기 때문에

옵저버 패턴 참고 : pjh3749.tistory.com/266

public interface Account { // observable
    void subscribe(Platform platform);
    void unSubscribe(Platform platform);
    void notifyCrew(String msg);
}

public interface Platform { // observer
    void update(String msg);
}

public class Jyami implements Account {
    List<Platform> platforms = new ArrayList<>();
    public void publishText(){
        System.out.println("쟈미가 글을 발행했다");
        notifyCrew("글 발행 전송");
    }
    @Override
    public void subscribe(Platform platform) {
        platforms.add(platform);
    }

    @Override
    public void unSubscribe(Platform platform) {
        platforms.remove(platform);
    }

    @Override
    public void notifyCrew(String msg) {
        platforms.forEach(crew -> crew.update(msg));
    }
}

public class Blog implements Platform {
    @Override
    public void update(String msg) {
        System.out.println("Blog 수신 : " + msg);
    }
}

public class Instagram implements Platform {
    @Override
    public void update(String msg) {
        System.out.println("instargram 수신 : "+ msg);
    }
}

public class Facebook implements Platform {
    @Override
    public void update(String msg) {
        System.out.println("FaceBook 수신 : " + msg);
    }
}

public class Main {
    public static void main(String[] args){
        Jyami jyami = new Jyami();
        Platform blog = new Blog();
        Platform facebook = new Facebook();
        Platform instagram = new Instagram();

        jyami.subscribe(blog);
        jyami.subscribe(facebook);
        jyami.subscribe(instagram);

        jyami.publishText();

        jyami.unSubscribe(facebook);

        jyami.publishText();
    }
}

쟈미가 글을 발행했다
Blog 수신 : 글 발행 전송
FaceBook 수신 : 글 발행 전송
instargram 수신 : 글 발행 전송
쟈미가 글을 발행했다
Blog 수신 : 글 발행 전송
instargram 수신 : 글 발행 전송

이렇게 보면 간단한데 플로 인터페이스의 개념을 복잡하게 만든 기능은 역압력과 압력이다.(스레드 활용에서 필수적)

압력 (push 모델) : 매초마다 수천개의 메세지가 onNext()로 제한없이 전달된다면? > 어느정도의 메세지가 들어올지 숫자를 제한하는 역압력과 같은 기법이 필요하다.

역압력 (pull 모델) :자바 9 Flow API에서 발행자가 무한의 속도로 아이템을 방출하는 대신, 요청했을 때만 다음 아이템을 보내도록 하는 request() 메서드를 제공(Subscription 인터페이스)

2. 역압력

Publisher와 Subscriber 사이에 채널이 연결되면 첫 이벤트로 Subscriber 인터페이스의 onSubscribe(Subscription subscription) 메서드가 호출된다.

이 Subscription 객체는 Subscriber와 Publisher가 통신할 수 있는 메서드를 포함한다 

옵저버 패턴과 pub-sub 패턴의 차이가 명확해 지는 부분은 이 subscription에 대한 부분 때문

3.  실제 역압력의 간단한 형태

한번에 한개의 이벤트를 처리하도록 pub-sub 연결을 구성하려면 다음과 같은 작업이 필요하다.

• Subscriber가 onSubscribe로 전달된 Subscription 객체를 subscription 같은 필드에 로컬로 저장한다.
• Subscriber가 수많은 이벤트를 받지 않게 onSubscribe, onNext, onError의 마지막 동작에 channel.request(1)을 추가해 오직 한 이벤트만 추가한다
• 요청을 보낸 체널에만 onNext, onError이벤트를 보내도록 Publisher의 notifyAllSubscribers 코드를 바꾼다. (보통 여러 Subscriber가 자신만의 속도를 유지할 수 있도록 Publisher는 새 Subscription을 만들어 각 Subscriber와 연결한다)

당김 기반 리액티브 역압력 : Subscriber가 Publisher로부터 요청을 당긴다(pull) = 리액티브 당김 기반(reactive pull based)라고 불린다.

6. 리액티브 시스템 vs 리액티브 프로그래밍

리액티브 시스템(reactive system) : 런타임 환경이 변화에 대응하도록 전체 아키텍처가 설계된 프로그램
www.reactivemanifesto.org/ : 공식 속성 문서

리액티브 시스템이 가져야할 속성 : 반응성(responsive), 회복성(resilient), 탄력성(elastic)
- 반응성 : 리액티브 시스템이 큰 작업을 하느라 간단한 질의의 응답을 지연하지 않고 실시간으로 입력에 반응한다.
- 회복성 : 한 컴포넌트의 실패로 전체 시스템이 실패하지 않는다.
- 탄력성 : 자신의 작업 부하에 맞게 적용하여 작업을 효율적으로 처리한다.

Java.util.concureent.Flow 관련된 자바 인터페이스에서 제공하는 리액티브 프로그래밍 형식을 이용하여 이 속성들을 구현할 수 있다.

자바 인터페이스 설계는 Reactive Manifesto의 마지막 속성인 메시지 주도(message-driven) 속성을 반영한다. > 박스와 채널모델

소스코드

https://github.com/mjung1798/Jyami-Java-Lab/tree/master/java8-in-action

동작 파라미터화 : 변화하는 요구사항에 효과적으로 대응할 수 있는 코드

• 익명 클래스 : 다양한 동작 구현 가능 -> 코드가 깔끔하지 않다
• 람다 표현식 : 더 깔끔한 코드로 동작을 구현할 수 있다.

1. 람다란 무엇인가

a. 람다의 특징

• 익명 : 보통 메서드와 다르게 이름이 없다
• 함수 : 메서드처럼 특정 클래스에 종속되지 않는다. 하지만 메서드처럼 파라미터 리스트 바디, 반환 형식, 가능한 예외 클래스 포함
• 전달 : 람다 표현식을 메서드 인수로 전달하거나 변수로 저장할 수 있다.
• 간결성 : 익명 클래스처럼 많은 자질구레한 코드를 구현할 필요가 없다.

b. 람다의 구성

Comparator<Apple> byWeight = (Apple a1, Apple a2) -> a1.getWeight.compareTo(a2.getWeight);

• 파라미터 리스트 : Comparator의 compare 메서드의 파라미터 (두 개의 사과)
• 화살표 : 람다의 파라미터 리스트와 바디를 구분
• 람다의 바디 : 두 사과의 무게를 비교한다. 람다의 반환값에 해당하는 표현식

c. 람다 예제

2. 람다를 사용하는 곳

함수형 인터페이스 문맥에서 사용

a. 함수형 인터페이스

• 정확히 하나의 추상메서드를 지정하는 인터페이스
• 많은 디폴트 메서드가 있더라도 추상 메서드가 오직 하나면 함수형 인터페이스

ex ) Comprator, Runnable

람다는 함수형 인터페이스의 추상 메서드 구현을 직접 전달 할 수 있어 전체 표현식을 함수형 인터페이스의 인스턴스로 취급할 수 있다.

@FunationalInterface : 함수형 인터페이스임을 가리키는 어노테이션, 이를 선언하고 실제로 함수형 인터페이스가 아니면 컴파일 에러 발생

b. 함수 디스크립터(function descriptor)

람다 표현식의 시그너처를 서술하는 메서드

() -> void : 파라미터 리스트가 없으며 void 를 반환하는 함수 : Runnable

(Apple, Apple) -> int : 두개의 Apple을 인수로 받아 int를 반환하는 함수

c. 사용법

• 함수형 인터페이스를 인수로 받는 메서드로 전달
• 변수에 할당

3. 람다 활용 : 실행 어라운드 패턴

실행 어라운드 패턴(execute around pattern) : 자원을 열고 -> 처리한 다음(변동) -> 자원을 닫는다

// 자원 열고 닫는 로직이 중복적으로 앞뒤에 있다.
public static String processFile2(BufferedReaderProcessor p) throws IOException {
  InputStream fileResourceAsStream = FileProcessor.class.getClassLoader().getResourceAsStream("data.txt");
  try(BufferedReader br = new BufferedReader(new InputStreamReader(fileResourceAsStream))){
    return p.process(br);
  }
}

//처리 로직에 대한 동작을 함수형 인터페이스를 이용해 파라미터화 하였다.
@FunctionalInterface
public interface BufferedReaderProcessor {
  String process(BufferedReader br) throws IOException;
}

// 처리 로직에 대한 동작을 선언하여, 실행 어라운드 패턴안에서 해당 동작을 수행한다.
@Test
@DisplayName("실행 어라운드 패턴 개선1")
void name2() throws IOException {
  String s = FileProcessor.processFile2(br -> br.readLine());
  assertThat(s).isEqualTo("line1");
}

4. 함수형 인터페이스 사용

함수 디스크립터(function descriptor) : 함수형 인터페이스의 추상 메서드 시그너처

공통의 함수 디스크립터를 기술하는 함수형 인터페이스 집합 : Comparable, Runnable, Callable 등 자바에서 포함하고 있다

a. Predicate

Predicate<String> lengthOnePredicate = (String s) -> s.length() == 1;
List<String> filter = PredicateProcess.filter(list, lengthOnePredicate);

• Input : 제네릭 형식 T 객체
• Output : Boolean 반환

b. Consumer

Consumer<Integer> printConsumer = i -> System.out.println(i);
ConsumerProcess.forEach(list, printConsumer);

• Input : 제네릭 형식 T 객체
• Output : void 반환

c. Function

Function<String, Integer> parseIntegerFunction = s -> Integer.parseInt(s);
List<Integer> map = FunctionProcess.map(list, parseIntegerFunction);

• Input : 제네릭 형식 T 객체
• Output : 제네릭 형식 R 객체

d. 기본형 특화

• 박싱(boxing) : 기본형 -> 참조형 변환
• 언박싱(unboxing) : 참조형 -> 기본형 변환
• 오토박싱(autoboxing) : 박싱과 언박싱이 자동으로 이루어짐

기본형 특화 함수형 인터페이스 : 기본형을 입출력으로 사용하는 상황에서 오토박싱 동작을 피할 수 있게 함

5. 형식 검사, 형식 추론, 제약

a. 형식검사

람다가 사용되는 컨텍스트를 이용해서 람다의 형식을 추론할 수 있다.

대상형식 (target type) : 어떤 컨텍스트에서 기대되는 람다 표현식의 형식

filter(inventory, (Apple a)->a.getWeight > 150);

1. 람다가 사용된 컨텍스트확인 : filter의 정의 확인
2. 두 번째 파라미터로 Predicate<Apple> 형식(대상 형식)을 기대한다.
3. Predicate<Apple>의 추상메서드를 확인한다.
4. Apple을 인수로 받아 boolean을 반환하는 test 메서드다 (Apple->boolean)
5. 함수 디스크립터와, 람다의 시그너처가 일치한다! : 형식검사 완료

b. 같은 람다 다른 함수형 인터페이스

대상 형식이라는 특징덕분에 같은 람다표현식이더라도 호환되는 추상메서드를 가진 다른 함수형 인터페이스 사용이 가능하다.

Callable<Integer> c = () -> 42;
PrivilegedAction<Integer> p = () -> 42;

하나의 람다표현식을 다양한 함수형 인터페이스에 사용할 수 있다.

c. 형식추론

자바 컴파일러

• 람다 표현식이 사용된 컨텍스트(대상 형식) 를 이용해 람다 표현식과 관련된 함수형 인터페이스 추론
• 대상 형식을 이용해서 함수 디스크립터를 알 수 있다.
• 컴파일러는 람다의 시그너처도 추론할 수 있다.

상황에 따라 명시적으로 형식을 포함하는 것이 좋을 때도 있고 형식을 배제하는 것이 가독성을 향상시킬 때도 있다.

// 형식 추론을 하지 않는다
Comparator<Apple> c = (Apple a1, Apple a2) -> a1.getWeight().compareTo(a2.getWeight());


// 형식 추론을 한다
Comparator<Apple> c = (a1, a2) -> a1.getWeight().compareTo(a2.getWeight());

d. 지역변수 사용

• 자유변수(free variable) : 파라미터로 넘겨진 변수가 아닌 외부에서 정의된 변수 => 람다캡처링(capturing lambda)
• 제약 : final로 선언되어있거나 실질적으로 final로 선언된 변수와 똑같이 사용되어야한다.

람다 표현식은 한번만 할당할 수 있는 지역변수를 캡쳐한다.

int portNumber = 1337;
Runnable r = ()->System.out.println(portNumber);
portNumber = 31337;

제약이 있는 이유

• 인스턴스 변수는 힙에 저장, 지역변수는 스택에 저장된다.
• 람다가 스레드에서 실행된다면 변수를 할당한 스레드가 사라져서 변수 할당이 해제되었는데도 람다를 실행하는 스레드에서 해당 변수에 접근하려 할 수 있다.
• 원래 변수에 접근을 허용하지 않고 자유 지역 변수의 복사본을 제공한다
• 복사본의 값이 바뀌지 않아야한다 == 지역 변수에는 한 번만 값을 할당해야 한다.

e. 클로저

클로저 : 함수의 비지역 변수를 자유롭게 참조할 수 있는 함수의 인스턴스

6. 메서드 레퍼런스

특정 메서드만을 호출하는 람다의 축약형이다.

명시적으로 메서드명을 참조하여 가독성을 높일 수 있다! 메서드 명 앞에 구분자(::)를 붙이는 방식으로 활용한다.

Apple 클래스에 정의된 getWeight 메서드 레퍼런스

• Apple::getWeight
• (Apple a)-> a.getWeight()

a. 메서드 레퍼런스를 만드는 방법

1. 정적 메서드 레퍼런스

Function<String, Integer> stringIntegerFunctionLambda = (String str) -> Integer.parseInt(str);
Function<String, Integer> stringIntegerFunction = Integer::parseInt;

2. 다양한 형식의 인스턴스 메서드 레퍼런스

Function<String, Integer> stringIntegerFunctionLambda = (String arg0) -> arg0.length();
Function<String, Integer> stringIntegerFunction = String::length;

3. 기존 객체의 인스턴스 메서드 레퍼런스

람다 표현식에 현존하는 외부 객체의 메서드를 호출 할 때 사용

Apple apple = new Apple("red", 100);
Supplier<Integer> supplierLambda = () -> apple.getWeight();
Supplier<Integer> supplier = apple::getWeight;

메서드 레퍼런스는 컨텍스트의 형식과 일치하는지 확인한다.

b. 생성자 레퍼런스

ClassName::new 처럼 클래스와 new 키워드를 사용해 기존 생성자의 레퍼런스를 만들 수 있다.

@Getter
public class Apple {
    private String color;
    private Integer weight;
    public Apple() {
    }
    public Apple(String color) {
        this.color = color;
    }
    public Apple(Integer weight) {
        this.weight = weight;
    }
    public Apple(String color, Integer weight) {
        this.color = color;
        this.weight = weight;
    }
}

// 기본 생성자 레퍼런스
Supplier<Apple> appleSupplier = Apple::new;

//인수 1개 생성자 레퍼런스 : 무게
Function<Integer, Apple> appleFunction = Apple::new;

// 인수 1개 생성자 레퍼런스 : 색깔
Function<String, Apple> appleFunction = Apple::new;

// 인수 2개 생성자 레퍼런스
BiFunction<String, Integer, Apple> appleBiFunction = Apple::new;

시그너처를 대응시켜서 생성자에 접근이 가능하다.

7. 람다, 메서드 레퍼런스 정리

• 코드 전달 : 함수형 인터페이스를 구현하여 사용
• 익명 클래스 사용 : 클래스를 구현하지 않고 바로 인스턴스 화 할 수 있으나 코드가 지저분하다.
• 람다 표현식 사용 : 추상 메서드의 시그너처(함수 디스크립터)가 람다 표현식의 시그너처를 정의한다 = 형식추론에 이용
• 메서드 레퍼런스 활용 : 람다 표현식의 인수를 더 깔끔하게 전달할 수 있다.

8. 람다 표현식을 조합할 수 있는 메서드

람다 표현식을 조합할 수 있는 유틸리티 메서드 : 디폴트메서드를 사용한다

@FunctionalInterface
public interface Comparator<T> {
  int compare(T o1, T o2); // 추상메서드
  default Comparator<T> reversed() {...}
  default Comparator<T> thenComparing(Comparator<? super T> other) {...}
}

실제로 위와 같이 선언이 되어있다.

단순한 람다 표현식을 조합해서 더 복잡한 람다 표현식을 만들 수 있다.

a. Comparator 조합

inventory.sort(
  Comparator.comparing(Apple::getWeight)
  .reversed() // 역정렬
  .thenComparing(Apple::getColor)); // 두번째 비교자를 만들 수 있다 (두 사과 비교 후 같을 때 정렬 법)

b. Predicate 조합

Predicate<Apple> redApple = a -> "red".equals(a.getWeight());
// 기존 Predicate를 반전
Predicate<Apple> notRedApple = redApple.negate();
// 기존 Predicate에 and를 이용해서 빨강이면서 무거운 사과로 조합
Predicate<Apple> redAndHeavyApple = redApple.and(a -> a.getWeight() > 150);
// 기존 Predicate에 or를 이용해서 빨강이면서 무거운 사과 또는 그냥 녹색사과로 조합
Predicate<Apple> redAndHeavyAppleOrGreen = redApple
  .and(a -> a.getWeight() > 150)
  .or(a -> "green".equals(a.getColor()));

c. Function 조합

andThen : 주어진 함수를 먼저 적용한 결과를 다른 함수의 입력으로 전달하는 함수를 반환

compose : 인수로 주어진 함수를 먼저 실행 한 후에 그 결과를 외부 함수의 인수로 제공

@Test
@DisplayName("Function 연결")
void name3() {
  Function<Integer, Integer> f = x -> x + 1;
  Function<Integer, Integer> g = x -> x * 2;
  Function<Integer, Integer> h = f.andThen(g); // g(f(x))
  int result = h.apply(1);
  assertThat(result).isEqualTo(4);
}

@Test
@DisplayName("Function 연결")
void name4() {
  Function<Integer, Integer> f = x -> x + 1;
  Function<Integer, Integer> g = x -> x * 2;
  Function<Integer, Integer> h = f.compose(g); // f(g(x))
  int result = h.apply(1);
  assertThat(result).isEqualTo(3);
}

소스코드

https://github.com/mjung1798/Jyami-Java-Lab/tree/master/java8-in-action

동작 파라미터화 : 아직은 어떻게 실행할 것인지 결정하지 않은 코드 블록, 코드 블록의 실행은 나중으로 미뤄진다.

변화하는 요구사항에 유연하게 대응할 수 있게 코드를 구현하는 방법

• 리스트의 모든 요소에 '어떤 동작'을 수행할 수 있음
• 리스트 관련 작업을 끝낸 다음에 '어떤 다른 동작'을 수행할 수 있음
• 에러가 발생하면 '정해진 어떤 다른 동작'을 수행할 수 있음

1. 변화하는 요구사항에 대응하기

DRY(don't repeat yourself) : 같은 것을 반복하지 말것.

문자열, 정수, 불린 등의 값으로 메서드를 파라미터화 할 때 => 한줄이 아니라 메서드 전체 구현을 바꾸어야 한다.

녹색 사과 필터링 filterGreenApples(List<Apple> inventory)
-> 색에 따른 필터링 filterApplyByColor(List<Apple> inventory, String color)
-> 속성 필터링 filterApples(List<Apple> inventory, String color, int wieght, boolean flag)

별로다

2. 동작 파라미터화

프레디케이트 Predicate : 선택 조건을 결정하는 인터페이스 => 어떤 속성에 기초해서 불린값을 반환

전략 디자인 패턴(strategy design pattern)

• 각 알고리즘(전략이라 불리는)을 캡슐화하는 알고리즘 패밀리를 정의해둔 다음에 런타임에 알고리즘을 선택하는 기법
• 객체가 할 수 있는 행위를 전략으로 만들고 동적으로 행위의 수정이 필요한 경우 전략으로 바꾸는 것만으로도 행위의 수정이 가능하도록 만든 패턴

public interface ApplePredicate{
  boolean test(Apple apple); 
}

public class AppleGreenColorPredicate implements ApplePredicate {
    @Override
    public boolean test(Apple apple) {
        return "green".equals(apple.getColor());
    }
}

public class AppleHeavyWeightPredicate implements ApplePredicate{
    @Override
    public boolean test(Apple apple) {
        return apple.getWeight() > 150;
    }
}

동작 파라미터화 : 메서드가 다양한 동작(전략)을 받아서 내부적으로 다양한 동작을 수행한다.

public static List<Apple> filterApples(List<Apple> inventory, ApplePredicate p){
  List<Apple> result = new ArrayList<>();
  for(Apple apple: inventory){
    if(p.test(apple))
      result.add(apple);
  }
  return result;
}

List<Apple> heavyApples = FilteringApple.filterApples(inventory, new AppleHeavyWeightPredicate());
List<Apple> greenApples = FilteringApple.filterApples(inventory, new AppleGreenColorPredicate());

내가 전달한 ApplePredicate 객체에 의해 filterApples 메서드의 동작이 결정된다.

한 메서드가 다른 동작을 수행하도록 재활용이 가능하다 : 한개의 파라미터 다양한 동작

3. 간소화 하기

a. 익명 클래스

익명클래스(anonymous class) 기법 : 클래스의 선언과 인스턴스화를 동시에 수행할 수 있는 기법

클래스 선언과 인스턴스화를 동시에 할 수 있다 == 즉석에서 필요한 구현을 만들어서 사용할 수 있다.

Predicate 인터페이스를 구현하는 클래스 만들어서 인스턴스 화 해야한다 (귀찮음)

List<Apple> redApples = FilteringApple.filterApples(inventory, new ApplePredicate() {
  @Override
  public boolean test(Apple apple) {
    return "red".equals(apple.getColor());
  }
});
assertThat(redApples.size()).isEqualTo(2);

그러나 장황하다 : 나쁘다!

람다 표현식으로 코드를 더 간결하게 정리할 수 있다.

b. 람다 표현식

List<Apple> redApples = FilteringApple.filterApples(inventory, apple -> "red".equals(apple.getColor()));

유연함과 간결함 두가지를 모두 가져간다!

c. 형식 파라미터를 이용한 추상화

public static <T> List<T> filter(List<T> list, Predicate<T> p){
  List<T> result = new ArrayList<>();
  for(T e: list){
    if(p.test(e))
      result.add(e);
  }
  return result;
}

d. 종합: 동작 파라미터화의 3가지 방법

• 클래스
• 익명 클래스
• 람다

5. 실전 예제

a. Comparator 정렬

sort 동작을 파라미터화 할 수 있다.

// 익명클래스 이용
inventory.sort(new Comparator<Apple>() {
  @Override
  public int compare(Apple o1, Apple o2) {
    return o1.getWeight().compareTo(o2.getWeight());
  }
});
// 람다식 이용
inventory.sort((a1, a2)->a1.getWeight().compareTo(a2.getWeight()));

b. Runnable 코드 블록 실행

// 익명클래스 이용
Thread t1 = new Thread(new Runnable() {
  @Override
  public void run() {
    System.out.println("helloWorld");
  }
});
t1.run();
// 람다식 이용
Thread t2 = new Thread(() -> System.out.println("helloWorld"));
t2.run();

소스코드

https://github.com/mjung1798/Jyami-Java-Lab/tree/master/java8-in-action

1. 자바의 변화

a. 자바의 시작

• 출발자체가 많은 유용한 라이브러리를 포함하는 잘 설계된 객체지향 언어
• 소소한 동시성도 지원
• JVM 바이트코드로 컴파일하는 특징 -> 인터넷 애플릿 프로그램 주 언어
• 캡슐화 덕분에 C에 비해 소프트웨어 엔지니어링적 문제가 훨씬 적다.

b. 언어 생태계의 변화

빅데이터에 직면하면서 멀티코어 컴퓨터, 컴퓨팅 클러스터를 이용하는 움직임이 생김 : 병렬 프로세싱 활용

대용량 데이터와 멀티코어 CPU를 효과적으로 활용해야 했다 => 자바 8의 변화

c. 자바 8의 3가지 프로그래밍 개념

1. 스트림 처리(Stream processing) : 여러행의 스트림을 입력으로 받아 여러 행의 스트림을 출력으로 만든다.

cat file1 file2 | tr "[A-Z]" "[a-z]" | sort | tail -3

Stream<T> : 하려는 작업을 추상화하여 일련의 스트림으로 만들어 처리 가능, 입력 부분을 여러 CPU 코어에 할당 가능

2. 동작 파라미터화(behavior prameterization)로 메서드에 코드 전달

코드의 일부를 API로 전달하는 기능 : 함수형 프로그래밍

3. 병렬성과 공유 가변 데이터

스트림 메서드로 전달하는 코드는 다른 코드와 동시에 실행되도 안전하게 실행이 된다.

공유 가변데이터에 접근하지 않기 때문이다 => 순수(pure) 함수, 부작용없는(side-effect-free) 함수, 상태없는(stateless) 함수 : 함수형 프로그래밍

자바 역시 하드웨어나 프로그래머의 기대의 변화에 부응할 수 있도록 변화하였다.

2. 자바 함수

자바의 함수 : 부작용을 일으키지 않는 함수

자바8로 인해 이급 시민 => 일급 시민 변화가 일어났다.

• 일급(first-class) 시민 : 조작이 가능한 것, 바꿀 수 있는 값 : 기본값(int, double), 객체(String, Integer, HashMap 등)
• 이급 시민 : 전달 할 수 없는 구조체, 바꿀 수 없는 값 : 메서드, 클래스

a. 메서드 레퍼런스(method reference)

이 메서드를 값으로 사용한다 : 메서드가 이급값이 아닌 일급값이 되었다.

객체 레퍼런스(new Object())로 객체를 주고받았던 것 처럼 메서드 레퍼런스를 만들어 전달 할 수 있게되었다.

b. 람다(lambda):익명함수(anonymous function)

함수도 값으로 취급할 수 있다.

직접 메서드를 정의할 수도 있지만, 이용할 수 있는 있는 편리한 클래스나 메서드가 없을 때 새로운 람다 문법을 이용하면 코드를 더 간결하게 구현할 수 있다.

람다 문법 형식으로 구현된 프로그램 == 함수형 프로그래밍 == 함수를 일급값으로 넘겨주는 프로그램을 구현한다.

3. 스트림(stream)

• 주어진 조건에 따라 데이터를 필터링(filtering)
• 데이터를 추출(extracting)
• 데이터 그룹화(gropuing)
• 동작을 쉽게 병렬화 가능
  포킹(forking) => 포킹된 리스트 처리 => 처리된 결과를 합침

추가 : 함수형 프로그래밍

• 함수를 일급값으로 사용한다
• 프로그램이 실행되는 동안 컴포넌트간에 상호작용이 일어나지 않는다(가변이 아니다)

4. 디폴트 메서드

기존 인터페이스의 변경에 대한 어려움 해결 : 인터페이스를 업데이트하면 인터페이스 구현클래스도 업데이트 해야했다.

구현 클래스에서 구현하지 않아도 되는 메서드를 인터페이스가 포함 할 수 있는 기능

5. 그 외

• Optional<T> 클래스 : 값을 갖거나 갖지 않을 수 있는 컨테이너 객체
• 구조적(structural) 패턴 매칭 기법 : 데이터 형식 분류와 분석을 한번에 수행

item 34. int 상수 대신 열거 타입을 사용하라

• 열거 타입은 확실히 정수 상수보다 뛰어나다. 더 알기쉽고 강력하다
• 대다수 열거 타입이 명시적 생성자나 메서드 없이 쓰이지만, 각 상수를 특정 데이터와 연결짓거나 상수마다 다르게 동작하게 할 때는 필요하다
• 하나의 메서드가 상수별로 다르게 동작해야할 때에는 switch문 대신 상수별 메서드 구현을 사용하자
• 열거 타입 상수 일부가 같은 동작을 공유한다면 전략 열거 타입 패턴을 사용하자.
• Link : jyami.tistory.com/102

item 35. ordinal 메서드 대신 인스턴스 필드를 사용하라

• ordinal 메서드는 EnumSet과 EnumMap과 같이 열거 타입 기반의 범용 자료 구조에 쓸 목적으로 설계 되었다. 이 경우가 아니면 사용하지 말자
• 열거 타입 상수에 연결 된 값을 ordinal 메서드로 얻지 말고 인스턴스 필드에 저장하라.
• Link : https://jyami.tistory.com/103

item 36. 비트 필드 대신 EnumSet을 사용하라

• 열거할 수 있는 타입을 한데 모아 집합 형태로 사용한다고 해도 비트 필드를 사용할 이유는 없다
• EnumSet 클래스가 비트 필드 수준의 명료함과 성능을 제공하고 아이템 34에서 설명한 열거 타입의 장점까지 선사한다
• EnumSet의 유일한 단점은 (자바 11까지는 아직) 불변 EnumSet을 만들 수 없다는 것이다.
• 향후 릴리즈에서 수정될 때 까지는 (명확성과 성능이 조금 희생되지만) Collections.unmodifiableSet으로 EnumSet을 감싸 사용할 수 있다.
• Link : https://jyami.tistory.com/104

item 37. ordinal 인덱싱 대신 EnumMap을 사용하라

• 배열의 인덱스를 위해 ordinal을 쓰는 것은 일반적으로 좋지 않으니 EnumMap을 사용하라
• 다차원 관계는 EnumMap<..., EnumMap<...>> 으로 표현하라
• 애플리케이션 프로그래머는 Enum.ordinal을 (왠만해서는) 사용하지 말아야한다 (아이템 35)는 일반원칙의 특수한 사례다
• Link : https://jyami.tistory.com/105

item 38. 확장할 수 있는 열거 타입이 필요하면 인터페이스를 사용하라

• 열거 타입 자체는 확장할 수 없지만, 인터페이스와 그 인터페이스를 구현하는 기본 열거 타입을 함께 사용해 같은 효과를 낼 수 있다.
• 이렇게 하면 클라이언트는 이 인터페이스를 구현해 자신만의 열거 타입(혹은 다른 타입)을 만들 수 있다.
• API가 (기본 열거 타입을 직접 명시하지 않고) 인터페이스 기반으로 작성되었다면 기본 열거 타입의 인스턴스가 쓰이는 모든 곳을 새로 확장한 열거 타입의 인스턴스로 대체해 사용 할 수 있다.
• Link : https://jyami.tistory.com/106

Item 39. 명명패턴보다 애너테이션을 사용하라

• 애너테이션으로 할 수 있는 일을 명명패턴으로 처리할 이유는 없다.
• 자바 프로그래머라면 예외 없이 자바가 제공하는 애너테이션 타입들은 사용해야한다.
• 마커애너테이션을 이용하여 애너테이션에 관심있는 프로그램에 추가 정보를 제공한다.
• 매개변수를 받는 애너테이션 타입, 반복가능한 애너테이션 등을 사용하여 소스 코드에 추가 정보를 제공할 수 있는 도구를 제공하자.
• Link : https://jyami.tistory.com/107

item 40. @Override 애너테이션을 일관되게 사용하라

• 재정의한 모든 메서드에 @Override 애너테이션을 의식적으로 달면 실수했을 때 컴파일러가 바로 알려준다.
• 예외는 하나다. 구체 클래스에서 상위 클래스의 추상 메서드를 재정의한 경우엔 이 애너테이션을 달지 않아도 된다. (단다고 해서 해롭진 않다.)
• Link : https://jyami.tistory.com/108

item 41. 정의하려는 것이 타입이라면 마커 인터페이스를 사용하라

• 마커 인터페이스와 마커 애너테이션은 각자의 쓰임이 있다.
• 새로 추가하는 메서드 없이 단지 타입 정의가 목적이라면 마커 인터페이스를 선택하자
• 클래스나 인터페이스의 프로그램요소에 마킹해야하거나 애너테이션을 적극 활용하는 프레임워크의 일부로 그 마커를 편입시키고자 한다면 마커 애너테이션이 올바른 선택이다.
• 적용 대상이 ElementType.TYPE인 마커 애너테이션을 작성하고 있다면, 잠시 여유를 갖고 정말 애너테이션으로 구현하는게 옳은지, 혹은 마커 인터페이스가 낫지는 않을지 곰곰히 생각해보자
• Link : https://jyami.tistory.com/109