Reactive Spring -- 1. Reactive概念

Spring 5 中引入了Reactive理念，下文主要介绍Reactive模式的基础。

工程地址，分支为reactive-operations。

Reactive概念

Reactive是函数式编程(Functional)，管道流(pipeline, stream), 异步非阻塞的。

org.reactivestreams包中主要有4个接口

• 发布者Publisher

public interface Publisher<T> {

    public void subscribe(Subscriber<? super T> s);
    
}

• 订阅者Subscriber

当接收到Publisher的数据时，会调用响应的回调方法。注册完成时，首先会调用onSubscribe方法，参数Subscription s包含了注册信息。

public interface Subscriber<T> {
    
    // 注册完成后，首先被调用
    public void onSubscribe(Subscription s);
    
    public void onNext(T t);
    
    public void onError(Throwable t);
    
    public void onComplete();
}

• 订阅Subscription

1. 通过订阅，订阅者Subscriber可以请求数据request，或者取消订阅cancel。
2. 在请求数据时，参数long n表示希望接收的数据量，防止发布者Publisher发送过多的数据。
3. 一旦开始请求，数据就会在流stream中传输。每接收一个，就会调用onNext(T t)；发生错误时，onError(Throwable t)被调用；传输完成后，onComplete()被调用。

public interface Subscription {
    
    // 请求数据，参数n为请求的数据量，不是超时时间
    public void request(long n);

    // 取消订阅
    public void cancel();
}

• Processor

可以看出，Processor接口继承了Subscriber和Publisher，是流的中间环节。

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {

}

Reactive Stream中数据从Publisher开始，经过若干个Processor，最终到达Subcriber，即完整的Pipeline。

Project Reactor

依赖

<dependency>
  <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core</artifactId>
</dependency>

Mono和Flux

1. 抽象类Mono和Flux实现了Publisher接口，他们是发布者。
2. Mono表示少于等于1个数据（即0个， 或1个数据）或错误；Flux表示一连串多个数据。

操作

1. 创建Flux或Mono，调用subscribe()后，数据开始流动。

主要方法有：just, fromArray, fromStream, fromIterable, range

    @Test
    public void create() {
    
        //just方法
        String[] arr = new String[]{"hello", "world"};
        Flux<String> flux1 = Flux.just(arr);
        flux1.subscribe(System.out::println);

        Mono<String> mono = Mono.just("hi world");
        mono.subscribe(System.out::println);
        
        //fromArray方法
        List<String> list = Arrays.asList("hello", "world");
        Flux<String> flux2 = Flux.fromIterable(list);
        
        //fromIterable方法
        List<String> fruitList = new ArrayList<>();
        fruitList.add("Apple");
        fruitList.add("Orange");
        fruitList.add("Grape");
        fruitList.add("Banana");
        fruitList.add("Strawberry");
        Flux<String> flux3 = Flux.fromIterable(fruitList);
        
        //fromStream方法
        Stream<String> stream = Stream.of("hi", "hello");
        Flux<String> flux4 = Flux.fromStream(stream);
        
        //range方法
        Flux<Integer> range = Flux.range(0, 5);
        
        //interval方法, take方法限制个数为5个
        Flux<Long> longFlux = Flux.interval(Duration.ofSeconds(1)).take(5);
    }
    
    

2. 合并mergeWith

@Test
public void mergeFlux() {
    Flux<String> source1 = Flux.just("hello", "world");
        Flux<String> source2 = Flux.just("hi", "ted");

        Flux<String> merge = source1.mergeWith(source2);
        merge.subscribe(System.out::println);
}

3. 结合为Tuple2元组类型zipWith

@Test
public void zipFlux() {
    Flux<String> source1 = Flux.just("hello", "world");
    Flux<String> source2 = Flux.just("hi", "ted");

    Flux<Tuple2<String, String>> zip = source1.zipWith(source2);
    zip.subscribe(tuple -> {
        System.out.println(tuple.getT1() + " -> " + tuple.getT2());
    });
}

4. 转换和过滤

skip: 略过2个

@Test
public void skipFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "hi", "ted");

    Flux<String> skip = source1.skip(2);
    skip.subscribe(System.out::println);
}

take：只取前2个

@Test
public void takeFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "hi", "ted");

    Flux<String> skip = source1.take(2);
    skip.subscribe(System.out::println);
}

filter: 接收Predicate

@Test
public void filterFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "hi", "ted");

    Flux<String> skip = source1.filter(s -> s.startsWith("h"));
    skip.subscribe(System.out::println);
}

distinct: 去重

@Test
public void distinctFlux() {
    Flux<String> source1 = Flux.just("hello", "hello", "world", "hi", "ted");

    Flux<String> skip = source1.filter(s -> s.startsWith("h")).distinct();
    skip.subscribe(System.out::println);
}

map: 接收Function

@Test
public void mapFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "hi", "ted");

    Flux<String> skip = source1.map(s -> s + " is mapped");
    skip.subscribe(System.out::println);
}

flatMap: 根据Flux中的元素先生成Mono, 再对Mono中的元素进行map转换。

@Test
public void flatMapFlux() {
    Flux<String> source1 = Flux.just("hello world", "hi ted");

    Flux<String> flatMap = source1.flatMap(s -> Mono.just(s).map(s1 -> {
        String[] strings = s1.split("\\s");
        return new String(strings[0] + " - " + strings[1]);
    }));

    flatMap.subscribe(System.out::println);
}

buffer: 将stream中的数据按照固定大小分配，新的Flux中的List的元素个数是2

@Test
public void bufferFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "hi", "ted");
    Flux<List<String>> buffer = source1.buffer(2);

    buffer.subscribe(strings -> System.out.println(strings.size()));
}

collectList: 将Flux中的元素收集到一个List中

@Test
public void collectListFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "hi", "ted");
    Mono<List<String>> mono = source1.collectList();
    
    mono.subscribe(System.out::println)
}

collectMap: 将Flux中的元素提取为一个Map，Map的key根据Function生成

@Test
public void collectMapFlux() {
        Flux<String> source1 = Flux.just("hello", "world", "ted");

        Mono<Map<Character, String>> map = source1.collectMap(s -> s.charAt(0));
        map.subscribe(characterStringMap -> System.out.println(characterStringMap.get('t')));
}

5. 逻辑运算

all: 判断Flux中元素是否都满足Predicate条件

@Test
public void allFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "ted");

    Mono<Boolean> mono = source1.all(s -> s.contains("e"));
    mono.subscribe(System.out::println);
}

any: 判断Flux中元素是否至少有1个满足Predicate条件

@Test
public void anyFlux() {
    Flux<String> source1 = Flux.just("hello", "world", "ted");

    Mono<Boolean> mono = source1.any(s -> s.contains("e"));
    mono.subscribe(System.out::println);
}