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

추천 튜토리얼: "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는 즉시 Future8742468051c85b06f0a0af9e3e506b5c를 반환합니다. 비동기 처리 프로세스는 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

또한 게으른 계산을 지원하지 않으며, 여러 값에 대한 지원이 부족하고, 고급 오류 처리 기능도 부족합니다.

명령형 프로그래밍에서 반응형 프로그래밍으로

Reactor와 같은 반응형 라이브러리의 목표는 JVM의 "전통적인" 비동기 접근 방식의 단점을 해결하는 동시에 몇 가지 추가 측면에도 초점을 맞추는 것입니다.

Can 구성 및 가독성.

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

합성 가능성 및 가독성

합성 가능성은 실제로 여러 비동기 작업을 조율하는 기능이므로 이전 작업의 결과를 후속 작업의 입력으로 사용하거나 포크-조인(포크-병합) 방식으로 수행할 수 있습니다. 여러 작업을 수행하거나 이러한 비동기 작업을 더 높은 수준에서 재사용할 수 있습니다.

작업을 조정하는 능력은 코드의 가독성 및 유지 관리 가능성과 밀접하게 연관되어 있습니다. 비동기 처리의 수와 복잡성이 증가함에 따라 코드를 작성하고 읽는 것이 더욱 어려워졌습니다.

보시다시피 콜백 모델은 간단하지만 콜백이 콜백 내에 중첩되어 여러 수준에 도달하면 콜백 지옥이 됩니다. 🎜🎜Reactor는 중첩 수준을 최소화하기 위한 풍부한 조합 옵션을 제공하므로 코드의 구성 구조는 어떤 종류의 추상화가 처리되고 있는지 반영할 수 있으며 일반적으로 동일한 수준으로 유지됩니다. 🎜🎜🎜조립 라인 비유🎜🎜🎜 마치 조립(생산) 라인을 통과하는 것처럼 데이터를 처리하는 반응형 애플리케이션을 생각할 수 있습니다. 리액터는 컨베이어 벨트이자 작업대입니다. 🎜🎜원자재는 소스(원 출판사)로부터 지속적으로 확보되어 완성된 제품이 소비자(구독자)에게 전달됩니다. 🎜🎜원자재는 다른 중간 단계 등 다양한 변형을 거치거나 대규모 조립 라인의 일부가 될 수 있습니다. 🎜🎜어딘가에 결함이나 막힘이 있는 경우 문제가 있는 워크스테이션은 업스트림에 알림을 보내 원자재의 흐름(속도)을 제한할 수 있습니다. 🎜🎜🎜운영자🎜🎜🎜Reactor에서 운영자는 조립 라인 비유에서 워크스테이션입니다. 각 연산자는 게시자에 몇 가지 동작을 추가하여 이전 게시자를 새 인스턴스로 래핑합니다. 이것이 전체 체인이 연결되는 방식입니다. 🎜🎜그래서 데이터는 처음에는 첫 번째 게시자로부터 나온 다음 체인을 따라 하류로 이동하며 각 링크별로 변환됩니다. 마지막으로 구독자는 프로세스를 종료합니다. 🎜🎜Reactive Stream 사양은 연산자를 명확하게 규정하지 않지만 Reactor는 간단한 변환 및 필터링부터 복잡한 오케스트레이션 및 오류 처리에 이르기까지 다양한 측면을 포함하는 풍부한 연산자를 제공합니다. 🎜🎜구독하지 않는 한 아무 일도 일어나지 않습니다🎜🎜퍼블리셔 체인을 작성하면 기본적으로 데이터가 체인에 입력되기 시작하지 않습니다. 대신 비동기 처리에 대한 추상적인 설명만 작성하면 됩니다. 🎜🎜이 동작(action)을 구독함으로써 게시자와 구독자가 연결되고, 이후 데이터가 트리거되어 체인에 흐르게 됩니다. 🎜🎜이는 내부적으로 구현되어 구독자의 요청 신호를 통해 업스트림으로 전파되어 원래 게시자까지 업스트림됩니다. 🎜🎜Reactor의 핵심 기능🎜🎜Reactor는 구성 가능한 반응형 유형을 도입하고 게시자 인터페이스를 구현하며 Flux 및 Mono와 같은 풍부한 연산자도 제공합니다. 🎜🎜Flux, 즉 흐름은 0부터 N까지의 요소를 나타냅니다. 🎜🎜Mono, 싱글은 0 또는 1개의 요소를 나타냅니다. 🎜🎜그들 사이의 차이점은 주로 의미론적이며 비동기 처리의 대략적인 기반을 나타냅니다. 🎜🎜http 요청이 하나의 응답만 생성하는 경우 이를 Monob9c6304980a6413006d7713f6edf94ae로 표현하는 것이 분명히 더 의미가 있으며 0/1과 같은 컨텍스트에 관련된 연산자만 제공합니다. 왜냐하면 카운트 작업은 당연히 별 의미가 없기 때문입니다. 이 시간에 . 🎜🎜 연산자는 처리의 최대 카디널리티를 변경할 수 있으며 관련 유형으로도 전환합니다. 예를 들어, 카운트 연산자가 Flux에 존재하더라도 그 반환 값은 Mono입니다. 🎜🎜🎜Flux8742468051c85b06f0a0af9e3e506b5c🎜🎜🎜A Flux8742468051c85b06f0a0af9e3e506b5c는 0~N 요소를 내보낼 수 있고 완료 신호나 오류 신호로 종료될 수 있는 비동기 시퀀스를 나타내는 표준 게시자8742468051c85b06f0a0af9e3e506b5c입니다. 🎜🎜반응형 스트리밍 사양과 마찬가지로 이 세 가지 유형의 신호는 다운스트림 구독자의 onNext, onComplete 및 onError 메서드에 대한 호출로 변환됩니다. 🎜🎜이 세 가지 방법은 이벤트/콜백으로도 이해될 수 있으며 모두 선택 사항입니다. 🎜🎜onNext가 없고 onComplete가 있는 경우 빈 유한 시퀀스를 나타냅니다. onNext나 onComplete가 없습니다. 이는 빈 무한 시퀀스를 나타냅니다(실용적이지 않으며 테스트에 사용할 수 있음). 🎜

무한 시퀀스는 비워둘 필요가 없습니다. 예를 들어 Flux.interval(Duration)은 무한하고 시계에서 방출되는 일반 "틱"인 Fluxad84a9ca022f0fe586e370cd3963bff2을 생성합니다.

Mono8742468051c85b06f0a0af9e3e506b5c

A Mono8742468051c85b06f0a0af9e3e506b5c는 최대 하나의 요소를 내보내고 onComplete 신호 또는 onError 신호를 사용하여 종료될 수 있는 특수 게시자8742468051c85b06f0a0af9e3e506b5c입니다.

제공하는 연산자는 Flux에서 제공하는 연산자의 하위 집합일 뿐입니다. 마찬가지로 일부 연산자(예: Mono와 Publisher 결합)도 Flux로 전환할 수 있습니다.

예를 들어 Mono#concatWith(Publisher)는 Flux를 반환하지만 Mono#then(Mono)는 다른 Mono를 반환합니다.

Mono는 Mono903bf37051cf83cbd9686768ac0189ae로 표현되는 반환 값(Runnable과 유사) 없는 비동기 처리를 나타내는 데 사용할 수 있습니다.

Flux 또는 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");

구독의 경우 Java 8 람다 표현식을 사용할 수 있으며 구독 방법에는 다양한 변형 본문이 있습니다. , 다른 콜백이 있습니다.

다음은 메소드 서명입니다.

//订阅并触发序列
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);

Disposable을 사용하여 구독 취소

이 람다 기반 구독 메소드는 모두 Disposable 유형을 반환하며 dispose()를 호출하여 구독을 취소할 수 있습니다.

Flux 및 Mono의 경우 취소는 소스가 요소 생성을 중단해야 한다는 신호입니다. 그러나 즉각적인 효과는 보장되지 않으며 일부 소스는 요소를 너무 빨리 생성하여 취소 신호가 수신되기 전에 완료될 수 있습니다.

추천 관련 기사: "Java 개발 튜토리얼"

위 내용은 자바 반응형 프로그래밍이란 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!