Java：Reactive Streams简单介绍

た入场券 2023-06-25 15:26 12阅读 0赞

Reactive Streams是一套规范，它为非阻塞带背压的异步流处理提供了一套标准。这套API主要提供了4个接口。它针对不同的开发语言都提供了API，为了讲解方便，下面主要以java为例进行介绍。  
以下是Reactive Streams官方提供的java API接口：

> <dependency>  
>     <groupId>org.reactivestreams</groupId>  
>     <artifactId>reactive-streams</artifactId>  
>     <version>1.0.3</version>  
> </dependency>

package org.reactivestreams;
    
    public interface Publisher<T> {
    
        public void subscribe(Subscriber<? super T> s);
    }

package org.reactivestreams;
    
    public interface Subscriber<T> {
    
        public void onSubscribe(Subscription s);
    
        public void onNext(T t);
    
        public void onError(Throwable t);
    
        public void onComplete();
    }

package org.reactivestreams;
    
    public interface Subscription {
    
        public void request(long n);
    
        public void cancel();
    }

package org.reactivestreams;
    
    public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
    }

java从JDK9开始，也提供了一套功能上完全等同于Reactive Streams的接口，java.util.concurrent.Flow

> java.util.concurrent.Flow.Processor<T,R>  
> java.util.concurrent.Flow.Publisher<T>  
> java.util.concurrent.Flow.Subscriber<T>  
> java.util.concurrent.Flow.Subscription

Reactive Streams官方介绍文档如下：  
[https://www.reactive-streams.org/][https_www.reactive-streams.org]  
Reactive Streams针对不同开发语言提供的API：  
[https://github.com/reactive-streams][https_github.com_reactive-streams]

由于Reactive Streams只是一套API接口规范，如果要实际使用它，需要对它进行实现。现在，很多项目对Reactive Streams进行了实现，其中比较有名的是ReactiveX、Project Reactor等，它们都提供了针对多种开发语言的实现版本。

ReactiveX的主要实现类：

<table> 
 <tbody> 
  <tr> 
   <td>io.reactivex.rxjava3.core.Flowable</td> 
   <td>0..N flows, supporting Reactive-Streams and backpressure</td> 
  </tr> 
  <tr> 
   <td>io.reactivex.rxjava3.core.Observable</td> 
   <td>0..N flows, no backpressure</td> 
  </tr> 
  <tr> 
   <td>io.reactivex.rxjava3.core.Single</td> 
   <td>a flow of exactly 1 item or an error</td> 
  </tr> 
  <tr> 
   <td>io.reactivex.rxjava3.core.Completable</td> 
   <td>a flow without items but only a completion or error signal</td> 
  </tr> 
  <tr> 
   <td>io.reactivex.rxjava3.core.Maybe</td> 
   <td>a flow with no items, exactly one item or an error</td> 
  </tr> 
 </tbody> 
</table>

Reactor的Publisher实现类：

<table style="width:680px;"> 
 <tbody> 
  <tr> 
   <td>reactor.core.publisher.Flux</td> 
   <td>an Asynchronous Sequence of 0-N Items</td> 
  </tr> 
  <tr> 
   <td>reactor.core.publisher.Mono</td> 
   <td>an Asynchronous 0-1 Result</td> 
  </tr> 
 </tbody> 
</table>

它们的主要资料可以在如下网址上查到：

ReactiveX：  
[http://reactivex.io/][http_reactivex.io]  
[https://github.com/ReactiveX/RxJava][https_github.com_ReactiveX_RxJava]

Project Reactor：  
[https://projectreactor.io/][https_projectreactor.io]

[https_www.reactive-streams.org]: https://www.reactive-streams.org/
[https_github.com_reactive-streams]: https://github.com/reactive-streams
[http_reactivex.io]: http://reactivex.io/
[https_github.com_ReactiveX_RxJava]: https://github.com/ReactiveX/RxJava
[https_projectreactor.io]: https://projectreactor.io/