Java中的Reactive編程示例
相信響應式編程經常會在各種地方被提到。本篇就為大家從函數式編程一直到Spring WeFlux做一次簡單的講解，并給出一些示例，希望大家可以更好的理解響應式編程，可以在合適的時機運用到實際項目中。
1. 前言
了解響應式編程，首先我們需要了解函數式操作和Stream的操作，下面我們簡單的復習一下嘍。
1.1 常用函數式編程
函數式接口中
我們先來回顧一下Java中的函數式接口。常見的有以下幾種
上面的簡單函數式接口示例如下：
Consumer consumer = (i)- > System.out.println("this is " + i);
consumer.accept("consumer");
Supplier supplier = () - > "this is supplier";
System.out.println(supplier.get());
Function< Integer,Integer > function = (i) - > i*i;
System.out.println(function.apply(8));
BiFunction< Integer,Integer,String > biFunction = (i,j)- > i+"*"+j+"="+i*j;
System.out.println(biFunction.apply(8,8));
Predicate< Integer > predicate = (i) - > i.intValue() >3;
System.out.println(predicate.test(5));
其執行結果如下：
this is consumer
this is supplier
64
8*8=64
true
1.2 Stream操作
對Stream進行操作，主要有幾個關鍵點：
- 生成流
- 流的中間操作 其中中間操作可以有多個，中間操作會返回一個新的流（如 map ,filter,sorted等），然后交給下一個流方法使用。
- 流的終結操作 終結操作只有一個。終結操作執行后，流就到了終止狀態，無法被操作 （如forEach，toArray , findFirst 等）。
創建流的示例：
String[] strArray = {"ss","ss","","sdffg"};
Arrays.stream(strArray).forEach(System.out::println);
Arrays.asList(strArray).stream().forEach(System.out::println);
Stream.of(strArray).forEach(System.out::println);
Stream.iterate(1,(i) - > i+1).limit(10).forEach(System.out::println);
Stream.generate(() - > new Random().nextInt(10)).limit(10).forEach(System.out::println);
簡單的流處理示例：
String[] strArray1 = {"ss","ss","","sdffg","bca-de","fff"};
String collect = Stream.of(strArray1)
.filter(i - > !i.isEmpty())//過濾空字符串
.sorted() //排序
.limit(1) //只取第一個元素
.map(i - > i.replace("-", ""))//替換 "-"
.flatMap(i - > Stream.of(i.split("")))//將字符拆成字符數組
.sorted() //排序
.collect(Collectors.joining());//將字符拼接組合到一起
System.out.println(collect);//最后輸出abcde
2. Java響應式編程
響應式編程會用到一個發布者和一個訂閱者，然后通過訂閱關系完成數據流的傳輸。訂閱關系中可以處理一些背壓問題，即調節消費者與生產者之間的供需平衡，讓整個程序達到最大效率。
Java9中java.util.concurrent.Flow接口提供響應式流編程類似的功能。
下面我們實現一個基于Java 響應式編程的示例：
其中有三個簡單步驟：
- 建立生產者
- 構建消費者
- 消費者訂閱生產者
- 生產者生產內容
SubmissionPublisher publisher = new SubmissionPublisher< >();//建立生產者
Flow.Subscriber subscriber = new Flow.Subscriber() {...};//建立消費者 (其中的實現放到下面)
publisher.subscribe(subscriber);//訂閱關系
for (int i = 0; i < 10; i++) {
publisher.submit("test reactive java : " +i); //生產者生產內容
}
消費者全部代碼如下：
Flow.Subscriber subscriber = new Flow.Subscriber() {
Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
System.out.println("Subscription establish first ");
this.subscription = subscription;
this.subscription.request(1);
}
@Override
public void onNext(Object item) {
subscription.request(10);
System.out.println("receive : "+ item);
}
@Override
public void onError(Throwable throwable) {
System.out.println(" onError ");
}
@Override
public void onComplete() {
System.out.println(" onComplete ");
}
};
其中onSubscribe方法表示建立訂閱關系
onNext接受數據，并請求生產者的數據。
onError，onComplete則是error或者完成之后的處理方法。
帶有中間處理器的響應式流
Reactive Stream 通常會基于如下的模型：
下面我們實現一個帶有中間處理功能的響應式模型：
下面的Processor 既有發布者，又有訂閱者：
public class ReactiveProcessor extends SubmissionPublisher implements Flow.Subscriber {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
System.out.println( Thread.currentThread().getName() + " Reactive processor establish connection ");
this.subscription = subscription;
this.subscription.request(1);
}
@Override
public void onNext(Object item) {
System.out.println(Thread.currentThread().getName() + " Reactive processor receive data: "+ item);
this.submit(item.toString().toUpperCase());
this.subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
System.out.println("Reactive processor error ");
throwable.printStackTrace();
this.subscription.cancel();
}
@Override
public void onComplete() {
System.out.println(Thread.currentThread().getName() + " Reactive processor receive data complete ");
}
}
如上中間處理器訂閱發布者， 同時消費者再訂閱中間處理器。中間處理器也可以調節發布訂閱的生產消費速率。
SubmissionPublisher publisher = new SubmissionPublisher< >(); //創建生產者
ReactiveProcessor reactiveProcessor = new ReactiveProcessor(); // 創建中間處理器
publisher.subscribe(reactiveProcessor); //中間處理器訂閱生產者
Flow.Subscriber subscriber = new Flow.Subscriber() {...}; //創建消費者
reactiveProcessor.subscribe(subscriber); //消費者訂閱中間處理器
for (int i = 0; i < 10; i++) {
publisher.submit("test reactive java : " +i); //生產者生產數據
}
通過上述生產者-> 中間處理器->消費者， 可以將生產者生產的數據全部變成大寫，然后再發送給最終的消費者。
以上式Java中的reactive 編程示例。Java會不同線程來分別處理消費者與生產者的消息處理
3. Reactor
Reactor中兩個比較關鍵的對象式Flux和Mono， 整個Spring的響應式編程均式基于projectreactor項目。Reactor是響應式編程的依賴，主要是基于JVM構建非阻塞程序。
根據Reactor的介紹，此類響應式編程的的三方庫（Reactor）主要是解決一些JVM經典異步編程中的一些缺點，并且還可以專注于一些新的特性，如下：
- 可組合性與可讀性 （Composability and readability）
- 可以使用豐富的運算操作符將數據作為流進行操作
- 訂閱之前，不會有任何事
- 背壓特性（Backpressure ），可以理解為消費者可以向生產者發送產出率過高的信號，從而調整生產速率。或者消費者可以選擇一次性拉去一捆數據進行消費。
- 于并發無關的高度抽象的高級功能
其中有這么一段解釋，可以形象的說明響應式編程。
Reactive的程序可以想象成車間的流水線，reactor既是流水線上的傳送帶，又是處理工作站。原料從一個原始的生產者出發，最終成為產品被推總給消費者。
3.1 Flux & Mono
下面我們介紹一下Flux和Mono。
在Reactor中Flux和Mono均是Publisher，即生產者。兩者也有不同。Flux對象表示0到N個異步的響應序列，而Mono只代表0個（empty）或者1個結果。
Reactor官網上介紹的Flux示意如下：
Mono示意如下：
3.2 Flux Mono創建與使用
我們也可以單獨引用其依賴。
使用maven依賴
< dependencies >
< dependency >
< groupId >io.projectreactor< /groupId >
< artifactId >reactor-core< /artifactId >
< /dependency >
< dependency >
< groupId >io.projectreactor< /groupId >
< artifactId >reactor-test< /artifactId >
< scope >test< /scope >
< /dependency >
< /dependencies >
Mono創建
分別創建空Mono和一個包含一個String的Mono，并由消費者消費打印。
Mono.empty().subscribe(System.out::println);
Mono.just("Hello Mono Java North").subscribe(System.out::print);
Flux創建
Flux創建有如下的一些方法，
- just（通過不定參數創建）
- range（從某個整數開始，往后的整數數量）
- fromArray，fromIterable，fromStream，從名稱上就可以看出來，通過數組，迭代器，Stream流創建Flux
下面式一些Java代碼示例
Flux.just(1,2,3,4,5).subscribe(System.out::print);
Flux.range(1,20).subscribe(System.out::print);
Flux.fromArray(new String[]{"a1","a2","a3","a4","a5","a6"}).skip(2).subscribe(System.out::print);
Flux.fromIterable(Arrays.asList(1,2,3,4,5,6,7)).subscribe(System.out::println);
Flux.fromStream(Stream.of(Arrays.asList(1,2,3,4,5,6,7))).subscribe(System.out::print);
我們再舉一個generate的例子
public static T, S > Flux< T > generate(Callable< S > stateSupplier, BiFunction< S, SynchronousSink< T >, S > generator)
如上代碼所示，generate需要一個Callable參數，而且是supplier （即沒有輸入值，只有一個輸出）
另一個參數是BiFunction （前面我們也介紹過，需要兩個輸入值，一個輸出值）。BiFunction中的其中一個輸入值是SynchronousSink，下面我們給出一個generate創建Flux的示例。
Flux.generate(
() - > 0, //提供一個初始狀態值0
(i, sink) - > {
sink.next("3*" + i + "=" + 3 * i);//使用初始值去生產一個3的乘法
if (i > 9) sink.complete();//設置停止條件
return i + 1;//返回一個新的狀態值，以便在下一次的生產中使用，除非響應序列終止
}).subscribe(System.out::println);
下面我們在看一個Flux嵌套處理示例：
需求：將字符串去空格，并去重，然后排序輸出。
String str = "qa ws ed rf tg yh uj i k ol p za sx dc vf bg hn jm k loi yt ";
Flux.fromArray(str.split(" "))//通過數組創建Flux
.flatMap(i - > Flux.fromArray(i.split("")))
.distinct() // 去重
.sort() //排序
.subscribe(System.out::print);
//flatMap與Stream中的flatMap類似，接受Function作為參數，輸入一個值，輸出一個值，此處輸出均為Publisher，
以上就是Flux和Mono的一些簡單介紹，同時Ractor也支持JDK中的FlowPubliser 和FlowSubscriber與 Reactor中的publisher, subscriber的適配等.
4. WebFlux
SpringBoot 2之后支持的Reactive響應式編程。
關于Reactive技術棧和經典的Servlet技術棧對比，Spring官網的這張圖比較清晰。
Spring響應式編程主要依賴于Reactor第三方庫，即上面講的Flux和Mono的庫。
WebFlux主要有以下幾個要點：
- 反應式棧web框架
- 完全異步非阻塞
- 運行在netty，undertow，Servlet3.1 + 容器
- 核心反應式庫 Reactor
- 返回 Flux 或Mono
- 支持注解和函數編程兩種編程模式
Spring WebFlux示例
下面我們給出幾個SpringBoot 的響應式web示例。
可以去https://start.spring.io/ 新建webflux的項目也可以。
項目中的主要依賴就是spring-boot-starter-webflux
< dependency >
< groupId >org.springframework.boot< /groupId >
< artifactId >spring-boot-starter-webflux< /artifactId >
< /dependency >
基于注解的WebFlux：
以下是一個最簡單的基于注解的WebFlux
@GetMapping("/hello/mono1")
public Mono< String > mono(){
return Mono.just("Hello Mono - Java North");
}
@GetMapping("/hello/flux1")
public Flux< String > flux(){
return Flux.just("Hello Flux","Hello Java North");
}
基于函數式編程的WebFlux:
創建RouterFunction，將其注入到Spring中即可。
@Bean
public RouterFunction< ServerResponse > testRoutes1() {
return RouterFunctions.route().GET("/flux/function", new HandlerFunction< ServerResponse >() {
@Override
public Mono< ServerResponse > handle(ServerRequest request) {
return ServerResponse.ok().bodyValue("hello web flux , Hello Java North");
}
}).build();
}
//上面的方法使用函數式編程替換之后如下
@Bean
public RouterFunction< ServerResponse > testRoutes() {
return RouterFunctions.route().GET("/flux/function",
request - > ServerResponse.ok()
.bodyValue("Hello web flux , Hello Java North")).build();
}
Flux與Mono的響應式編程延遲示例
下面我們編寫一段返回Mono的響應式Web服務。
@GetMapping("/hello/mono")
public Mono< String > stringMono(){
Mono< String > from = Mono.fromSupplier(() - > {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "Hello, Spring Reactive date time:"+ LocalDateTime.now();
});
System.out.println( "thread : " + Thread.currentThread().getName()+ " === " + LocalDateTime.now() +" ==========Mono function complete==========");
return from;
}
使用postman請求如下， 5秒鐘后返回數據。后臺卻在5秒中之前已經處理完整個方法。
后臺打印日志：
再看一組Flux
@GetMapping(value = "/hello/flux", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux< String > flux1(){
Flux< String > stringFlux = Flux.fromStream(IntStream.range(1,6).mapToObj(i - >{
mySleep(1);//表示睡1秒
return "java north flux" + i + "date time: " +LocalDateTime.now();
}));
System.out.println("thread : " + Thread.currentThread().getName()+ " === " + LocalDateTime.now() + " ==========Flux function complete=========");
return stringFlux;
}
此次使用谷歌瀏覽器請求此服務：
可以發現每隔一秒就會有一條消息被生產出來。
后臺完成時間同樣是在一開始就完成整個方法：
通過上述對Flux 與 Mono的例子，可以好好體會一下響應式編程。
總結
本篇回顧了函數式編程，Stream操作等，然后再舉例講了Java中的Reactive編程示例， 同時也給處理Reactor三方庫的Flux于Mono的示例。
最后使用了SpringBoot WebFlux 創建簡單的響應式web服務。希望能讓大家更好的理解響應式編程。
-
接口
+關注
關注
33文章
8526瀏覽量
150863 -
JAVA
+關注
關注
19文章
2960瀏覽量
104565 -
編程
+關注
關注
88文章
3596瀏覽量
93610 -
函數
+關注
關注
3文章
4308瀏覽量
62445
發布評論請先 登錄
相關推薦
工商網監
評論