做调研用到的大数据网站,做网站推广前途,一流的常州网站建设,做二手车网站怎么做的Java的反应式流是一种新的编程模型#xff0c;它在异步和事件驱动的环境下工作。反应式流的目的是为了解决传统的单线程或者多线程编程模型在高并发和大流量情况下的性能瓶颈。
反应式流的核心是Observable和Observer#xff0c;Observable表示一个数据流#xff0c;而Obse…Java的反应式流是一种新的编程模型它在异步和事件驱动的环境下工作。反应式流的目的是为了解决传统的单线程或者多线程编程模型在高并发和大流量情况下的性能瓶颈。
反应式流的核心是Observable和ObserverObservable表示一个数据流而Observer则表示这个数据流的消费者。Observable在数据流上产生事件而Observer则对这些事件进行响应。反应式流的数据流是一种推式的流Observable发布事件时不需要等待Observer接收Observable会把事件推送给Observer而不是Observer去轮询Observable。
Java的反应式流通常基于Reactor或RxJava等库这些库提供了丰富的函数式编程API和运算符可以非常方便地处理异步事件。这些库都提供了类似于Observable和Observer的抽象概念可以用来描述和处理异步数据流。同时还提供了常用的运算符包括map、filter、reduce等这些运算符可以方便地对数据流进行变换和过滤。
反应式流还有一个重要的概念是背压(backpressure)它是指在高并发和大流量情况下消费者无法处理生产者产生的数据流导致数据积压的情况。为了解决这个问题反应式流引入了背压机制生产者会在发送数据前先询问消费者的处理能力如果消费者没有处理能力生产者会等待一段时间或者缓存数据等待消费者处理完数据后再继续发送。
反应式流已经被广泛应用于大规模的互联网应用中包括机器学习、数据分析、网络爬虫等领域。它的优点在于处理高并发和大流量的数据流时能够更加高效地利用系统资源提高系统的性能和可扩展性。
总之反应式流是Java编程中的一个重要概念它可以帮助我们更好地处理异步和事件驱动的数据流提高系统的性能和可扩展性。 不涉及任何库就单纯用java的反应式流完成发布订阅者模式
package com.example.jdk9.react;import java.util.concurrent.Flow.*;public class PublisherSubscriberDemo {public static void main(String[] args) {SimplePublisherString publisher new SimplePublisher();SimpleSubscriberString subscriber1 new SimpleSubscriber();SimpleSubscriberString subscriber2 new SimpleSubscriber();publisher.subscribe(subscriber1);publisher.subscribe(subscriber2);publisher.submit(hello);publisher.submit(world);publisher.close();}
}class SimplePublisherT implements PublisherT {private Subscription subscription;Overridepublic void subscribe(Subscriber? super T subscriber) {subscriber.onSubscribe(new Subscription() {Overridepublic void request(long n) {}Overridepublic void cancel() {// nothing to do}});this.subscription new Subscription() {private boolean cancelled false;Overridepublic void request(long n) {// nothing to do}Overridepublic void cancel() {this.cancelled true;}public boolean isCancelled() {return this.cancelled;}};subscriber.onSubscribe(this.subscription);}public void submit(T item) {subscriptionLimitedQueue.offer(item);subscription.request(1);}public void close() {while (!subscriptionLimitedQueue.isEmpty()) {subscriptionLimitedQueue.poll();}subscription.cancel();}private SubscriptionLimitedQueueT subscriptionLimitedQueue new SubscriptionLimitedQueue(2);static class SubscriptionLimitedQueueT {private final int limit;private int size 0;private NodeT head;private NodeT tail;public SubscriptionLimitedQueue(int limit) {this.limit limit;}private static class NodeT {final T item;NodeT next;Node(T item, NodeT next) {this.item item;this.next next;}}public void offer(T item) {NodeT node new Node(item, null);if (head null) {head node;tail head;} else {tail.next node;tail tail.next;}size;if (size limit) {NodeT newHead head.next;head.next null;head newHead;size--;}}public boolean isEmpty() {return size 0;}public T poll() {if (isEmpty()) {return null;}T item head.item;NodeT newHead head.next;head.next null;head newHead;size--;return item;}}
}class SimpleSubscriberT implements SubscriberT {private Subscription subscription;Overridepublic void onSubscribe(Subscription subscription) {this.subscription subscription;System.out.println(订阅成功);subscription.request(1);}Overridepublic void onNext(T item) {System.out.println(Received item: item);subscription.request(1);}Overridepublic void onError(Throwable throwable) {throwable.printStackTrace();}Overridepublic void onComplete() {System.out.println(Done);}
}
这段代码演示了使用Flow API来发布和订阅消息的过程它包含以下类和接口
PublisherT发布者接口表示能够发布指定类型的消息给订阅者。SubscriberT订阅者接口表示能够接收指定类型的消息。Subscription订阅接口表示订阅关系能够请求一定数量的消息和取消订阅。SubmissionPublisherT继承自PublisherT接口实现了异步发布消息的能力。Flow API一组用于处理数据流和异步操作的接口和类。
具体解释
SimplePublisher类是一个实现了Publisher接口的简单发布者类它能够发布指定类型的消息给订阅者。它内部维护了一个SubscriptionLimitedQueue类的对象用于限制消息队列的长度。SubscriptionLimitedQueue类是一个维护队列长度的类用于实现限制消息队列长度的功能。SimpleSubscriber类是一个实现了Subscriber接口的简单订阅者类它能够接收指定类型的消息并将其输出到控制台中。main方法创建了一个SimplePublisher类的实例和一个SimpleSubscriber类的实例然后将它们关联起来最后向SimplePublisher类的实例中发布了两个消息随后关闭了发布者。
运行结果 例子
第一步引入依赖
dependencygroupIdio.projectreactor/groupIdartifactIdreactor-core/artifactIdversion3.5.11/version/dependency
第二步编写代码
package com.example.jdk9.react;import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;public class ReactiveStreamExample {public static void main(String[] args) {FluxInteger stream Flux.range(1, 10);stream.map(i - i * 2).filter(i - i % 3 0).flatMap(i - Mono.just(i).zipWith(Mono.just(i * 3))).subscribe(System.out::println);}
}
上面的代码首先创建了一个从1到10的数字列表然后通过map操作符将每个元素乘以2再使用filter操作符过滤掉不能被3整除的元素。接下来使用flatMap操作符来创建一个新的流该流将原始元素和该元素乘以3的结果合并在一起。最后使用subscribe方法来订阅这个流并打印出每个元素的值。
这个例子展示了Reactor库中的一些常见操作符包括map、filter和flatMap。通过这些操作符的链式调用我们可以轻松地对数据流进行复杂的操作。在实际的应用中我们可以根据具体的需求选择不同的操作符来实现所需的数据处理逻辑。 使用Reactor 库实现发布订阅者模式
package com.example.jdk9.react;import reactor.core.publisher.Flux;public class PublisherSubscriberExample {public static void main(String[] args) {// 创建发布者FluxInteger publisher Flux.just(1, 2, 3, 4, 5);// 订阅者1打印每个元素publisher.subscribe(System.out::println);// 订阅者2计算元素的总和并打印publisher.reduce(0, Integer::sum).subscribe(total - System.out.println(Sum total));}
}
