这篇文章主要介绍“java9新特性Reactive Stream响应式编程api怎么用”,在日常操作中,相信很多人在java9新特性Reactive Stream响应式编程API怎么用问题上存在疑惑,小编查阅了各式资料,
这篇文章主要介绍“java9新特性Reactive Stream响应式编程api怎么用”,在日常操作中,相信很多人在java9新特性Reactive Stream响应式编程API怎么用问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”java9新特性Reactive Stream响应式编程API怎么用”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
Java 9提供了一组定义响应式流编程的接口。所有这些接口都作为静态内部接口定义在java.util.concurrent.Flow
类里面。
下面是Java 响应式编程中的一些重要角色和概念,先简单理解一下
发布者(Publisher)是潜在的无限数量的有序数据元素的生产者。 它根据收到的需求(subscription)向当前订阅者发布一定数量的数据元素。
订阅者(Subscriber)从发布者那里订阅并接收数据元素。与发布者建立订阅关系后,发布者向订阅者发送订阅令牌(subscription),订阅者可以根据自己的处理能力请求发布者发布数据元素的数量。
订阅令牌(subscription)表示订阅者与发布者之间建立的订阅关系。 当建立订阅关系后,发布者将其传递给订阅者。 订阅者使用订阅令牌与发布者进行交互,例如请求数据元素的数量或取消订阅。
public static interface Subscriber<T> { public void onSubscribe(Subscription subscription); public void onNext(T item); public void onError(Throwable throwable); public void onComplete();}
onSubscribe
:在发布者接受订阅者的订阅动作之后,发布任何的订阅消息之前被调用。新创建的Subscription
订阅令牌对象通过此方法传递给订阅者。
onNext
:下一个待处理的数据项的处理函数
onError
:在发布者或订阅遇到不可恢复的错误时调用
onComplete
:当没有订阅者调用(包括onNext()方法)发生时调用。
订阅令牌对象通过Subscriber.onSubscribe()
方法传递
public static interface Subscription { public void request(long n); public void cancel();}
request(long n)
是无阻塞背压概念背后的关键方法。订阅者使用它来请求n个以上的消费项目。这样,订阅者控制了它当前能够接收多少个数据。cancel()
由订阅者主动来取消其订阅,取消后将不会在接收到任何数据消息。
@FunctionalInterfacepublic static interface Publisher<T> { public void subscribe(Subscriber<? super T> subscriber);}
调用该方法,建立订阅者Subscriber与发布者Publisher之间的消息订阅关系。
处理者Processor 可以同时充当订阅者和发布者,起到转换发布者——订阅者管道中的元素的作用。用于将发布者T类型的数据元素,接收并转换为类型R的数据并发布。
public static interface Processor<T,R> extends Subscriber<T>, Publisher<R> {}
现在我们要去实现上面的四个接口来完成响应式编程
Subscription Interface
订阅令牌接口通常不需要我们自己编程去实现,我们只需要在知道request()方法和cancle()方法含义即可。
Publisher Interface
发布者接口,Java 9 已经默认为我们提供了实现SubmissionPublisher,该实现类除了实现Publisher接口的方法外,提供了一个方法叫做submit()来完成消息数据的发送。
Subscriber Interface
订阅者接口,通常需要我们自己去实现。因为在数据订阅接收之后,不同的业务有不同的处理逻辑。
Processor
实际上是 Publisher Interface和Subscriber Interface的集合体,有需要数据类型转换及数据处理的需求才去实现这个接口
下面的例子实现的式字符串的数据消息订阅处理
import java.util.concurrent.Flow;public class MySubscriber implements Flow.Subscriber<String> { private Flow.Subscription subscription; //订阅令牌 @Override public void onSubscribe(Flow.Subscription subscription) { System.out.println("订阅关系建立onSubscribe: " + subscription); this.subscription = subscription; subscription.request(2); } @Override public void onNext(String item) { System.out.println("item: " + item); // 一个消息处理完成之后,可以继续调用subscription.request(n);向发布者要求数据发送 //subscription.request(n); } @Override public void onError(Throwable throwable) { System.out.println("onError: " + throwable); } @Override public void onComplete() { System.out.println("onComplete"); }}
import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Flow;import java.util.concurrent.SubmissionPublisher;public class SubmissionPublisherExample { public static void main(String[] args) throws InterruptedException { ExecutorService executor = Executors.newFixedThreadPool(1); SubmissionPublisher<String> sb = new SubmissionPublisher<>(executor, Flow.defaultBufferSize()); sb.subscribe(new MySubscriber()); //建立订阅关系,可以有多个订阅者 sb.submit("数据 1"); //发送消息1 sb.submit("数据 2"); //发送消息2 sb.submit("数据 3"); //发送消息3 executor.shutdown(); }}
控制台打印输出结果
订阅关系建立
onSubscribe: java.util.concurrent.SubmissionPublisher$BufferedSubscription@27e81a39
item: 数据 1
item: 数据 2
请注意:即使发布者submit了3条数据,MySubscriber也仅收到了2条数据进行了处理。是因为我们在MySubscriber#onSubscribe()
方法中使用了subscription.request(2);
。这就是“背压”的响应式编程效果,我有能力处理多少数据,就会通知消息发布者给多少数据。
到此,关于“java9新特性Reactive Stream响应式编程API怎么用”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注编程网网站,小编会继续努力为大家带来更多实用的文章!
--结束END--
本文标题: java9新特性Reactive Stream响应式编程API怎么用
本文链接: https://www.lsjlt.com/news/324826.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
2024-05-24
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
0