[모던 자바 인 액션] 9주차

[jeongminkyo]

스터디 날짜
2023.08.25 금 9:00-10:00

내용
챕터17. 리액티브 프로그래밍
챕터18. 함수형 관점으로 생각하기

공유

최승위

• https://www.loom.com/share/a499f12c87f94fae8fd23c2fdfc9736b?sid=807994c9-4666-461a-8b2f-66044a96967a
• https://www.loom.com/share/ef5c508180c144c5b217171a3665086a?sid=5f94dc5d-b26c-4f0c-9cf4-c87293bffb37

이성온

• https://www.loom.com/share/f8af3a4ee273414e9f2041c39e81e9f3?sid=405ef5d6-091a-4c3c-9f39-14f3503c50d3

정민교

• https://www.loom.com/share/0bc9c180f8094bf08c17cb7d00c62e80?sid=a11d170c-de6e-4058-9cf0-b36aa8c2fd56

[jeongminkyo]

시스템 구현과 유지보수

공유된 가변 데이터

변수가 예상하지 못하는 값을 갖는 이유는 시스템의 여러 메서드에서 공유된 가변 데이터 구조를 읽고 갱신하기 때문이다. 어떤 자료구조도 바꾸지 않는 시스템이 있다면 문제가 일어날 일이 없다(유지 보수하기 쉽다).

공유된 가변 데이터 구조로 인해 변수가 예상하지 못한 값을 갖게 되고 데이터 갱신 사실을 추척하기 어려워진다.

자신을 포함하는 클래스의 상태 그리고 다른 객체의 상태를 바꾸지 않으며 return 문을 통해서만 자신의 결과를 반환하는 메서드를 순수(pure) 메서드 또는 부작용 없는(side-effect free) 메서드라고 부른다. 이때 말하는 부작용은 아래와 같다.

• 자료구조를 고치거나 필드에 값을 할당(setter)
• 예외 발생
• 파일에 쓰기 등의 I/O 동작 수행

부작용 없는 시스템 컴포넌트에서는 메서드가 서로 간섭하지 않으므로 잠금을 사용하지 않고도 멀티코어 병렬성을 사용할 수 있다.

함수형 프로그래밍이 제공하는 부작용 없음(no side effect)과 불변성(immutability)이라는 개념은 시스템의 유지보수에 큰 도움을 준다.

선언형 프로그래밍

'어떻게(how)' 구현할 것인지 집중하는 프로그래밍 형식은 고전적 객체지향 프로그래밍에서 이용하는 방식이다. 명령형 프로그래밍이라고 부르기도 한다.

Transaction mostExpensive = transactions.get(0);
if (mostExpensive == null)
  throw new IllegalArgumentException("Empty list of transactions");

for(Transaction t: transactions.subList(1, transactions.size())) {
  if (t.getValue() > mostExpensive.getValue()) {
    mostExpensive = t;
  }
}

'무엇을' 에 집중하는 방식도 있다. 선언형 프로그래밍이라고 부르며, 우리가 원하는 것이 무엇이고 시스템이 어떻게 그 목표를 달성할 것인지 등의 규칙을 정한다.

Optional<Transaction> mostExpensive =
  transactions.stream().max(comparing(Transaction::getValue));

함수형 프로그래밍

• 함수형 프로그래밍은 말 그대로 함수를 이용하는 프로그래밍.

• 함수형 프로그래밍에서 말하는 함수는 수학적인 함수와 같다.

• 함수는 0개 이상의 인수를 가지며 한 개 이상의 결과를 반환하지만 부작용이 없어야 한다.

• 순수 함수형 프로그래밍 - '함수 그리고 if-then-else 등의 수학적 표현만 사용'하는 방식

• 함수형 프로그래밍 - 시스템의 다른 부분에 영향을 미치지 않는다면 내부적으로는 함수형이 아닌 기능도 사용하는 방식

함수형 자바

• 자바에서는 순수 함수형이 아닌 함수형 프로그래밍을 구현.

• 함수나 메서드는 지역변수만을 변경해야 하며, 참조하는 객체가 있다면 불변객체여야 한다.

• 메서드 내에서 생성한 객체의 필드는 갱신할 수 있지만, 외부에 노출되어서는 안된다.

• 함수나 메서드가 어떤 예외도 일으키지 않아야 한다. 예외가 발생하면 return으로 결과를 반환할 수 없게 되기 때문이다.

• 어떤 수를 0으로 나눈다든가 하는 함수의 경우 예외가 필요할 수 있다. 이럴 경우 Optional을 사용하면 예외 없이도 연산을 성공적으로 수행했는지 확인할 수 있다.

• 비함수형 동작을 감출 수 있는 상황에서만 부작용을 포함하는 라이브러리 함수를 사용해야 한다.

[so3500]

9주차 리액티브 프로그래밍

💡 모던 자바 인 액션 > 17장 내용 정리

변화하는 상황

• 빅 데이터 : 페타바이트 단위로 구성되며 매일 증가
• 다양한 환경 : 모바일 디바이스에서 클라우드 기반 클러스터에 이르는 다양한 환경
• 사용 패턴 : 사용자는 1년 내내 서비스를 이용할 수 있으며 밀리초 단위의 응답 시간을 기대

리액티브 프로그래밍은 높은 응답성과 가용성을 제공하여 이러한 문제를 해결한다.

리액티브 매니패스토

• 반응성(responsive) : 빠르면서 일정하고 예상할 수 있는 반응시간을 제공한다.
• 회복성(resilient) : 장애가 발생해도 시스템 반응성이 유지된다.
  • 컴포넌트 실행 복제, 여러 컴포넌트의 시간과 공간분리, 각 컴포넌트가 비동기적으로 작업을 다른 컴포넌트에 위임
• 탄력성(elastic) : 무거운 작업 부하 발생 시 자동으로 컴포넌트에 할당된 자원 수를 늘려 반응성을 유지한다.
• 메시지 주도(Message-driven) : 약한 결합, 고립, 위치 투명성을 제공하며 비동기 메시지를 전달해 컴포넌트끼리 통신하며 회복성과 탄력성을 얻는다.
  • 장애를 메시지로 처리 → 회복성
  • 주고받은 메시지의 수를 감시하고 메시지 양에 따라 리소스 할당 → 탄력성

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

• 이벤트 스트림을 블록하지 않고 비동기로 처리하는 것이 최신 멀티코어 CPU 사용률을 극대화할 수 있는 방법이다.
• 이를 위해 상대적으로 비싸고 희귀한 자원인 스레드를 퓨처, 액터, 콜백을 발생시키는 이벤트 루프등과 공유하고 처리할 이벤트를 변환하고 관리한다.
• 이들 기술은 스레드보다 가벼우며 비동기 애플리케이션 구현의 추상수준을 높일 수 있다.
• 블록, 경쟁조건, 데드락 같은 저수준의 멀티스레드 문제를 직접 처리할 필요가 없어지면서 비즈니스 요구사항 구현에 더 집중할 수 있다.
• 메인 이벤트 루프 안에서는 절대 동작을 블락하지 않아야 한다는 전제조건에 따른다. (데이터베이스나 파일 시스템 접근, 원격 서비스 호출 I/O 관련 동작)
  • RxJava, Akka 같은 리액티브 프레임워크는 별도 지정된 스레드 풀에서 블록 동작을 실행시켜 이 문제를 해결한다.
• 비교적 짧은 시간에 유지되는 데이터 스트림에 기반한 연산을 수행하며, 보통 이벤트 주도로 분류된다.
• ⇒ 시간에 기반한 비동기, 동시적 비결합

시스템 수준의 리액티브

• 여러 애플리케이션이 한 개의 일관적인 회복할 수 있는 플랫폼을 구성할 수 있게 해준다.
• 애플리케이션 중 하나가 실패해도 전체 시스템이 계속 운영될 수 있도록 한다.
• 애플리케이션을 조립하고 상호소통을 조절한다. (메시지 주도)
• 컴포넌트에서 발생한 장애를 고립시킴으로써 문제가 다른 곳으로 전파되지 않도록 막는다 (회복성, 결함 허용능력)
• 시스템의 모든 컴포넌트가 수신자의 위치에 상관없이 다른 모든 서비스와 통신할 수 있다. 이러한 위치 투명성 덕분에 시스템을 복제할 수 있으며 현재 작업 부하에 따라 애플리케이션을 자동 확장할 수 있다. (탄력성)
• ⇒ 위치 투명성을 통한 공간적 비결합

리액티브 스트림과 플로 API

• 리액티브 스트림은 잠재적으로 무한한 비동기 데이터를 순서대로, 블록하지 않는 역압력을 전제해 처리하는 표준 기술이다.
• 역압력(backpressure) : 발행-구독 프로토콜에서 이벤트 스트림의 구독자가 이벤트를 제공하는 속도보다 느린 속도로 이벤트를 소비하면서 문제가 발생하지 않도록 보장하는 장치
  • 부하가 발생한 컴포넌트는 이벤트 발생 속도를 늦추라고 알리거나, 얼마나 많은 이벤트를 수신할 수 있는지 알리거나, 다른 데이터를 받기 전에 기존의 데이터를 처리하는 데 얼마나 시간이 걸리는지를 업스트림 발행자에게 알릴 수 있어야 한다.

Flow 클래스

• 자바9에서 리액티브 프로그래밍을 제공하기 위해 추가한 클래스 (java.util.concurrent.Flow)
• 정적 컴포넌트 하나를 포함하고 있으며 인스턴스화할 수 없음
• 4개의 인터페이스 제공 : Publisher, Subscriber, Subscription, Processor

public final class Flow {

	private Flow() {
	} // uninstantiable

	// 구독자가 받는 메시지를 발행하는 생산자
	@FunctionalInterface
	public static interface Publisher<T> {
		// 주어진 구독자를 등록
		public void subscribe(Subscriber<? super T> subscriber);
	}

	// 메시지를 받는 구독자
	// onSubscribe onNext* (onError | onComplete)?
	// onSubscribe - onNext - onNext - onNext - onComplete
	public static interface Subscriber<T> {
		// Subscriber 가 Publisher 에 자신을 등록할 때 Publisher 가 onSubscribe 를 호출해 Subscription을 전달한다.
		public void onSubscribe(Subscription subscription);

		// 메시지를 수신하면 호출
		public void onNext(T item);

		// 메시지를 정상적으로 받지 못할 경우 호출
		public void onError(Throwable throwable);

		// 추가 메시지가 없을 경우 호출
		public void onComplete();
	}

	// Publisher 와 Subscriber 간 메시지를 제어하며 Subscriber 에 의해서만 호출된다.
	public static interface Subscription {
		// publishser 에게 주어진 메시지를 처리할 준비가 되었음을 알림
		public void request(long n);

		// publisher 에게 이벤트를 받지 않음을 통지
		public void cancel();
	}

	// 리액티브 스트림에서 처리하는 이벤트의 변환단계를 나타낸다.
	// Publisher 를 구독한 다음 수신한 데이터를 가공해 다시 제공할 때 사용한다.
	public static interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
	}

	static final int DEFAULT_BUFFER_SIZE = 256;

	// Publisher 와 Subscriber 의 기본 버퍼 사이즈를 제공한다.
	public static int defaultBufferSize() {
		return DEFAULT_BUFFER_SIZE;
	}

}

인터페이스 사용 규칙

• Publisher 는 반드시 Subscription 의 request 메서드에 정의된 개수 이하만 Subscriber에 전달해야 한다.
• Subscriber는 요소를 받아 처리할 수 있음을 Publisher에게 알려야 한다. (역압력 행사)
• Publisher와 Subscriber는 정확하게 Subscription을 공유해야 하며 각각이 고유한 역할을 수행해야 한다. 그러려면 onSubscribe와 onNext 메서드에서 Subscriber는 request 메서드를 동기적으로 호출할 수 있어야 한다.

그림 17-3 플로 API를 사용하는 리액티브 애플리케이션의 생명주기

[s5646s]

https://reactivex.io/
https://github.com/ReactiveX/RxJava
https://github.com/ReactiveX/rxjs
https://rxviz.com/examples/custom

const { interval } = Rx;
const { filter, take, map } = RxOperators;
 
interval(500).pipe(
  	filter(n => n % 2 == 0),
  	take(5),
  	map(n => Math.pow(n,2))
 )

const { fromEvent } = Rx;
const { filter, take, map } = RxOperators;

fromEvent(document, 'click').pipe(
  map(e => e.x),
  filter(x => x < 400),
	take(5)
)

const { fromEvent } = Rx;
const { delay, timeInterval, pluck, debounceTime } = RxOperators;

fromEvent(document, 'click').pipe(
	//delay(100)
  //timeInterval(), pluck('interval')
  debounceTime(1000)
)

<script src="https://unpkg.com/@reactivex/rxjs/dist/global/rxjs.umd.js"></script>

const { ajax } = rxjs.ajax
const { map } = rxjs.operators

ajax('https://jsonplaceholder.typicode.com/todos/1').pipe(
  map(result => result.response.title)
).subscribe(console.log)