자바 반응형 프로그래밍이란 무엇입니까?

추천 튜토리얼: "java 비디오 튜토리얼"

Java 반응형 프로그래밍이란 무엇인가요?

java 리액티브 프로그래밍은

리액티브 프로그래밍

리액티브 프로그래밍 방향의 첫 번째 단계로 Microsoft는 .NET 생태계에서 Rx 라이브러리(Reactive Extensions)를 만들었습니다. RxJava는 JVM에서의 구현입니다.

리액티브 프로그래밍은 일반적으로 Observer 패턴의 확장으로 객체 지향 언어에 나타나는 비동기 프로그래밍 패러다임입니다.

데이터의 흐름과 변화의 전파에 중점을 둡니다. 이는 정적(예: 배열) 또는 동적(예: 이벤트 이미터) 데이터 흐름을 프로그래밍 언어를 사용하여 쉽게 표현할 수 있음을 의미합니다.

Reactive Streaming

시간이 지나면서 Java 전용 표준화가 등장했습니다. JVM 플랫폼의 반응형 라이브러리에 대한 일부 인터페이스 및 상호 작용 규칙을 정의하는 사양입니다. 이는 Reactive Streams이며 해당 인터페이스는 Java 9의 상위 클래스 java.util.concurrent.Flow에 통합되었습니다. 반응형 스트림은 반복자와 유사하지만 반복자는 "풀"을 기반으로 하고 반응형 스트림은 "푸시"를 기반으로 합니다. 반복자의 사용은 실제로 명령형 프로그래밍입니다. 왜냐하면 다음 요소를 얻기 위해 next()를 호출할 시점을 결정하는 것은 개발자의 몫이기 때문입니다. 반응형 스트리밍에서 위와 동등한 것은 게시자-구독자입니다. 그러나 새로운 요소를 사용할 수 있게 되면 게시자가 이를 구독자에게 푸시합니다. 이 "푸시"가 응답성의 핵심입니다.

게다가 푸시된 요소에 대한 작업도 선언적 방식으로 수행됩니다. 프로그래머는 무엇을 해야 할지 표현하기만 하면 되며 어떻게 해야 할지 걱정할 필요가 없습니다.

게시자는 onNext 메소드를 사용하여 구독자에게 새 요소를 푸시하고, onError 메소드를 사용하여 오류를 알리고, onComplete 메소드를 사용하여 종료되었음을 알립니다.

오류 처리와 완료(종료)도 잘 처리되는 것을 볼 수 있습니다. 오류와 끝 모두 시퀀스를 종료할 수 있습니다.

이 방법은 매우 유연합니다. 이 모드는 요소 0(없음) / 요소 1개 / 요소 n(다)의 경우를 지원합니다(시계가 똑딱거리는 경우 무한 시퀀스 포함).

Reactor 출시

Reactor는 4세대 반응형 라이브러리이자 JVM 플랫폼의 반응형 스트림 사양을 기반으로 비차단 비동기 애플리케이션을 구축하는 데 사용되는 반응형 프로그래밍 패러다임의 구현입니다.

JVM(http://www.reactive-streams.org/)에서 반응형 스트림 사양을 크게 구현합니다.

효율적인 수요 관리("배압" 관리 형태)를 갖춘 완전한 비차단 반응형 프로그래밍의 초석입니다.

Java 기능 API, 특히 CompletableFuture, Stream 및 Duration을 직접 통합합니다.

Reactor-Netty 엔지니어링을 사용하여 비차단 프로세스 간 통신을 지원하고 마이크로서비스 아키텍처에 적합하며 HTTP(웹소켓 포함), TCP 및 UDP를 지원합니다.

참고: Reactor에는 Java 8+가 필요합니다.

너무 많이 말했지만, 먼저 왜 그러한 비동기 반응 라이브러리가 필요한지 생각해 봐야 합니까?

차단은 낭비입니다

최신 애플리케이션은 매우 많은 수의 동시 사용자에 도달할 수 있습니다. 최신 하드웨어의 기능이 지속적으로 향상되더라도 최신 소프트웨어의 성능은 여전히 ​​주요 관심사입니다

일반적으로 두 가지 방법이 있습니다. 프로그램 성능 향상:

1. 더 많은 스레드와 더 많은 하드웨어 리소스를 사용하는 병렬화

2. 효율성 향상, 현재 리소스 사용량에서 더 높은 효율성 추구

일반적으로 Java 개발자는 차단 코드를 사용하여 프로그램을 작성합니다. 이는 성능 병목 현상이 발생할 때까지 잘 작동합니다.

이번에는 유사한 차단 코드를 실행하기 위한 추가 스레드가 도입될 예정입니다. 그러나 이러한 확장 방식은 논란의 여지가 있을 수 있으며 리소스 활용 측면에서 동시성 문제로 이어질 수 있습니다.

더 나쁜 것은 차단은 자원을 낭비한다는 것입니다. 주의 깊게 살펴보면 프로그램에 약간의 지연(특히 데이터베이스 요청이나 네트워크 호출과 같은 I/O)이 포함되면 스레드가 이제 데이터를 기다리며 유휴 상태이므로 리소스가 낭비됩니다.

그래서 병렬화는 만능이 아닙니다. 하드웨어가 최대의 성능을 발휘할 수 있도록 해야 하지만, 자원 낭비의 영향과 원인도 매우 복잡합니다.

비동시성 구조

앞서 언급한 두 번째 방법은 더 높은 효율성을 추구하는 것인데, 이는 자원 낭비 문제의 해결책이 될 수 있습니다.

비동기식 비차단 코드를 작성하면 동일한 기본 리소스를 사용하여 실행을 다른 활성 작업으로 전환하고 나중에 현재 프로세스로 돌아갈 수 있습니다.

하지만 JVM에 대한 비동기 코드를 생성하는 방법은 무엇입니까? Java는 두 가지 비동기 프로그래밍 모델을 제공합니다.

1. 비동기 메서드에는 반환 값이 없지만 결과가 제공되면 콜백이 호출됩니다.

2. 비동기 방식인 Futures는 즉시 Future를 반환합니다. 비동기 처리 프로세스는 T 값을 계산하고 이에 대한 액세스를 래핑하기 위해 Future 개체를 사용하는 것입니다. 이 값은 즉시 사용할 수 없으며 개체를 폴링하여 T 값을 사용할 수 있는지 확인할 수 있습니다.

두 기술 모두 충분히 좋은가요? 모든 상황에 완벽하지는 않으며 두 방법 모두 제한 사항이 있습니다.

콜백은 함께 구성하기 어렵고 읽기 및 유지 관리가 어려운 코드(잘 알려진 "콜백 지옥")로 빠르게 이어질 수 있습니다.

콜백에 비해 Future는 약간 더 좋지만 여전히 조합에서는 잘 작동하지 않습니다. 여러 Futures 객체를 결합하는 것은 가능하지만 쉽지는 않습니다.

Future에는 다른 문제도 있습니다. get() 메서드 호출로 인해 또 다른 차단이 발생하기 쉽습니다. Future也有其它问题，很容易因为调用了get()方法造成了另一个阻塞。

另外，它不支持延迟计算，缺乏对多个值的支持，缺乏高级错误处理。

从命令式到响应式编程

像Reactor这样的响应式库的目标就是解决在JVM上“传统”异步方式的弊端，同时也关注一些额外方面：

可组合性和可读性。

数据作为流，被丰富的操作符操作，什么都不会发生，直到你订阅，后压，消费者通知生产者发射的速率太快了，高级别而不是高数值抽象。

可组合性和可读性

可组合性，其实就是编排多个异步任务的能力，使前一个任务的结果作为后续任务的输入，或以fork-join（分叉-合并）的方式执行若干个任务，或在更高的级别重复利用这些异步任务。

任务编排的能力和代码的可读性和可维护性紧密地耦合在一起。随着异步处理在数量和复杂度上的增加，组合和阅读代码变得更加困难。

就像我们看到的，回调模型虽然简单，但是当回调里嵌套回调，达到多层时就会变成回调地狱。

Reactor提供丰富的组合选项，使嵌套级别最小，让代码的组织结构能反映出在进行什么样的抽象处理，且通常保持在同级别上。

装配线类比

你可以认为响应式应用处理数据就像通过一个装配（生产）线。Reactor既是传送带又是工作站。

原材料从一个源（原始发布者）持续不断地获取，以一个完成的产品被推送给消费者（订阅者）结束。

原材料可以经过许多不同的转换，如其它的中间步骤，或者是一个更大装配线的一部分。

如果在某个地方出现一个小故障或阻塞了，出问题的工作站可以向上游发出通知来限制原材料的流动（速率）。

操作符

在Reactor里，操作符就是装配线类比中的工作站。每一个操作符都向一个发布者添加某些行为，把上一步的发布者包装到一个新的实例里。整个链就是这样被链接起来的。

所以数据一开始从第一个发布者出来，然后沿着链往下游移动，且被每一个链接转换。最后，一个订阅者结束了这个处理。

响应式流规范并没有明确规定操作符，不过Reactor就提供了丰富的操作符，它们涉及到很多方面，从简单的转换、过滤到复杂的编排、错误处理。

只要不订阅，就什么都不发生

当你写一个发布者链时，默认，数据是不会开始进入链中的。相反，你只是创建了异步处理的一个抽象描述。

通过订阅这个行为（动作），才把发布者和订阅者连接起来，然后才会触发数据在链里流动。

这是在内部实现好的，通过来自于订阅者的request信号往上游传播，一路逆流而上直到最开始的发布者那里。

Reactor核心特性

Reactor引入可组合响应式的类型，实现了发布者接口，但也提供了丰富的操作符，就是Flux和Mono。

Flux，流动，表示0到N个元素。

Mono

풍부한 운영자가 조작하는 스트림으로서의 데이터는 구독할 때까지 아무 일도 일어나지 않으며, 게시 후 소비자는 생산자에게 방출 속도가 너무 빠르며 높은 수치 추상화보다는 높은 수준임을 알립니다.

합성 가능성 및 가독성

Flux<String> seq1 = Flux.just("foo", "bar", "foobar");
List<String> iterable = Arrays.asList("foo", "bar", "foobar");
Flux<String> seq2 = Flux.fromIterable(iterable);
Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);
Mono<String> noData = Mono.empty();
Mono<String> data = Mono.just("foo");

//订阅并触发序列
subscribe(); 
//可以对每一个产生的值进行处理
subscribe(Consumer<? super T> consumer); 
//还可以响应一个错误
subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer); 
//还可以在成功结束后执行一些代码
subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer); 
//还可以对Subscription执行一些操作
subscribe(Consumer<? super T> consumer,
          Consumer<? super Throwable> errorConsumer,
          Runnable completeConsumer,
          Consumer<? super Subscription> subscriptionConsumer);