Reactor知识

本文主要是介绍响应式异步编程库Reactor的使用
响应式流简介

When the publisher is faster than the subscriber, the latter must have an unbounded buffer to store fast incoming items or it must drop items it cannot handle.Another solution is to use a strategy called backpressure in which the subscriber tells the publisher to slow down and hold the tems until the subscriber is ready to process more. Using backpressure may require the publisher to have an unbounded buffer if it keeps producing and storing elements for slower subscribers.The publisher may implement a bounded buffer to store a limited number of elements and may choose to drop them if its buffer is full.

What does the subscriber do when it requests items from the publisher and the items are not available?In a synchronous request, the subscriber must wait, possibly indefinitely, until items are available. If the publisher sends items to the subscriber synchronously and the subscriber processes them synchronously, the publisher must block until the data processing finishes. The solution is to have an asynchronous processing at both ends, where the subscriber may keep working on other tasks after requesting items from the publisher. When more items are ready, the publisher sends them to the subscriber asynchronously.

Reactive Streams started in 2013 as an initiative for providing a standard for asynchronous stream processing with non-blocking backpressure. It is aimed at solving the problems of processing a stream of items—how do you pass a stream of items from a publisher to a subscriber without requiring the publisher to block or the subscriber to have an unbounded buffer or drop.

The Reactive Streams model is very simple—the subscriber sends an asynchronous request to the publisher for N items. The publisher sends N or fewer items to the subscriber asynchronously.

Reactive Streams dynamically switches between the pull model and the push model streamprocessing mechanisms. It uses the pull model when the subscriber is slower and uses the push model when the subscriber is faster.

Reactor介绍
webflux与webmvc的类比:

webmvc webflux
controller handler
request mapping router
   <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-core</artifactId>
        <version>3.1.4.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId>
        <version>3.1.4.RELEASE</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>

一,Flue和Mono的简单用法

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;


public class SimpleReactor 
{
    private static void testConstructUsingJust()
    {
        //subscribe方法中的lambda表达式作用在了每一个数据元素上
        Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::print);
        System.out.println();//回车的作用
        Mono.just(1).subscribe(System.out::println);
    }
    
    private static void testConstructFromArray()
    {
        Integer[] array = new Integer[]{1,2,3,4,5,6};
        Flux.fromArray(array).subscribe(x -> {
            System.out.println("收到 "+ x);
            });     
    }
    
    private static void testConstructFromList()
    {
        List<Integer> list = Arrays.asList(1,2,3,4,5,6);
        Flux<Integer> flux = Flux.fromIterable(list);   
        flux.subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("Completed!"));
    }
    
    private static void testConstructFromStream()
    {
        Stream<Integer> stream = Arrays.asList(1,2,3,4,5,6).stream();
        Flux.fromStream(stream).subscribe(System.out::print);
    }
    
    private  static void testMonoError()
    {
        Mono.error(new Exception(" 注意注意,发生异常,注意处理啦")).subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("Completed!")
        );
    }
    
    public static void main( String[] args )
    {
        testConstructUsingJust();
        nextTest();
        testConstructFromArray();
        nextTest();
        testConstructFromList();
        nextTest();
        testMonoError();
        nextTest();
        testConstructFromStream();
    }
    
    private static void nextTest()
    {
        System.out.println("********************************");
    }
}

二、Reactor中如何做單元測試


import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

public class SimpleReactorTest
{
    private Flux<Integer> generateFluxFrom1To6()
    {
        return Flux.just(1, 2, 3, 4, 5, 6);
    }

    private Mono<Integer> generateMonoWithError()
    {
        return Mono.error(new Exception("some error"));
    }

    @Test
    public void testViaStepVerifier()
    {
        StepVerifier.create(generateFluxFrom1To6()).expectNext(1, 2, 3, 4, 5, 6)
                .expectComplete().verify();
        StepVerifier.create(generateMonoWithError())
                .expectErrorMessage("some error").verify();
    }
}

三、Flux和Mono也支持map、flatMap、Filter、zip等operator

flatMap示意图

@Test
    public void testMapAndFlatMap()
    {
        // 注意下面的6表示6個,和IntStream的Range方法里面不一样
                StepVerifier.create(Flux.range(1, 6).map(i -> i * i))
                        .expectNext(1, 4, 9, 16, 25, 36).expectComplete();

        StepVerifier
                .create(Flux.just("flux", "mono")
                        .flatMap(s -> Flux.fromArray(s.split("\\s*"))
                                .delayElements(Duration.ofMillis(100)))
                        .doOnNext(System.out::print))
                .expectNextCount(8).verifyComplete();
    }

    @Test
    public void testFilter()
    {
        StepVerifier.create(Flux.range(1, 6).filter(i -> i % 2 == 1) // 1
                .map(i -> i * i)).expectNext(1, 9, 25) // 2
                .verifyComplete();
    }

    private Flux<String> getZipDescFlux()
    {
        String desc = "Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.";
        return Flux.fromArray(desc.split("\\s+")); // 1
    }

    @Test
    public void testZip() throws InterruptedException
    {       
        CountDownLatch countDownLatch = new CountDownLatch(1);
        //使用Flux.interval声明一个每200ms发出一个元素的long数据流;因为zip操作是一对一的,故而将其与字符串流zip之后,字符串流也将具有同样的速度;
        Flux.zip(getZipDescFlux(), Flux.interval(Duration.ofMillis(200)))
                .subscribe(
                        t -> System.out.println(t.getT1()), 
                        null,
                        countDownLatch::countDown); // 4
        countDownLatch.await(10, TimeUnit.SECONDS); // 5
    }

有些内容还没有研究完,请接着看 http://blog.51cto.com/liukang/2090191或者https://www.ibm.com/developerworks/cn/java/j-cn-with-reactor-response-encode/index.html

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • rljs by sennchi Timeline of History Part One The Cognitiv...
    sennchi阅读 7,449评论 0 10
  • PLEASE READ THE FOLLOWING APPLE DEVELOPER PROGRAM LICENSE...
    念念不忘的阅读 13,552评论 5 6
  • 我一直十分感慨太多人没有思想性。盲从不思考本质用英雄联盟这个游戏举个例子 上路皇子打瑞文,两人等级均为六级。瑞文血...
    皮皮鲁与鲁西西阅读 643评论 0 0
  • 温老师,如果住在赵春江工作室,请准备以下物品: 一、毛巾洗漱用品,被子还挺干净的,没有筷子,可以带点一次性筷子,最...
    浅浅淡淡唯真阅读 458评论 0 0
  • 今天的情绪失控完全在意料之外。在帮你搬箱子下去的时候还依然平静,还在跟你开着玩笑。却在关上后备箱的一刹那有点想哭...
    木木姑娘阅读 223评论 0 1