Spring-Reactor 学习

响应式编程

什么是响应式编程

响应式编程（reactive programming）是一种基于数据流（data stream）和变化传递（propagation of change）的声明式（declarative）的编程范式。

响应式编程的好处

响应式编程是异步非阻塞的，能够有效的利用服务器资源，提高性能。它提升的并不是单个API的响应时间，而是提升整体服务并发处理量，默认Tomcat有200个线程同时只能处理200个请求，而基于数据流和事件循环的响应式框架能更好的利用CPU和内存资源。

Screen Shot 2020-10-22 at 5.35.25 PM.png

Spring Reactor

设计原理

观察者模式的延伸，Push模型

核心接口

Publisher
public interface Publisher<T> {

    // 传入订阅者
    public void subscribe(Subscriber<? super T> s);
}
Subscriber
public interface Subscriber<T> {
    
    // 注册完成后，首先被调用
    public void onSubscribe(Subscription s);
    // 执行消费函数
    public void onNext(T t);
    // Error时调用
    public void onError(Throwable t);
    // 完成订阅后执行
    public void onComplete();
}
Subscription
public interface Subscription {
    
    // 请求数据，参数n为请求的数据量，不是超时时间
    public void request(long n);
    // 取消订阅
    public void cancel();
}
Processor
public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {

}

BackPressure

通过Subscription接口实现BackPressure处理，调用request方法调整上游传递的数据量，默认是Long.MAX_VALUE。

Screen Shot 2020-10-22 at 5.47.19 PM.png

核心接口关系

  • Publisher调用subscribe方法，接受Subscriber对象参数。
  • 在subscribe方法中，Publisher调用Subscriber对象的onSubscribe方法，传入Subscription对象。
  • 在Subscription由Subscribe和Publisher中的数据组成（以ArraySubscription实现类为例）。
  • 在Subscription中根据request的值对Publisher中的数据进行循环调用Subscribe的onNext方法。
    @Test
    void fluxTest() {
        Flux<Integer> range = Flux.range(1, 5)
                .log();

        range.subscribe();
    }
17:49:40.157 [Test worker] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
17:49:40.159 [Test worker] INFO reactor.Flux.Range.1 - | request(unbounded)
17:49:40.160 [Test worker] INFO reactor.Flux.Range.1 - | onNext(1)
17:49:40.160 [Test worker] INFO reactor.Flux.Range.1 - | onNext(2)
17:49:40.160 [Test worker] INFO reactor.Flux.Range.1 - | onNext(3)
17:49:40.160 [Test worker] INFO reactor.Flux.Range.1 - | onNext(4)
17:49:40.160 [Test worker] INFO reactor.Flux.Range.1 - | onNext(5)
17:49:40.160 [Test worker] INFO reactor.Flux.Range.1 - | onComplete()

    @Test
    void fluxBaseSubscriber() {
        Flux<Integer> ints = Flux.range(1, 4)
                .log();
        ints.subscribe(new BaseSubscriber<>() {
            int count = 0;

            public void hookOnSubscribe(Subscription subscription) {
                System.out.println("Subscribed");
                request(2);
            }

            public void hookOnNext(Integer value) {
                System.out.println(value);
                if (++count == 2) {
                    request(2);
                }
            }
        });
    }
17:51:50.405 [Test worker] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
Subscribed
17:51:50.408 [Test worker] INFO reactor.Flux.Range.1 - | request(2)
17:51:50.408 [Test worker] INFO reactor.Flux.Range.1 - | onNext(1)
1
17:51:50.408 [Test worker] INFO reactor.Flux.Range.1 - | onNext(2)
2
17:51:50.408 [Test worker] INFO reactor.Flux.Range.1 - | request(2)
17:51:50.409 [Test worker] INFO reactor.Flux.Range.1 - | onNext(3)
3
17:51:50.409 [Test worker] INFO reactor.Flux.Range.1 - | onNext(4)
4
17:51:50.409 [Test worker] INFO reactor.Flux.Range.1 - | onComplete()

Hot and Cold

Cold

订阅前什么都不会发生，发布者进行了两次订阅，每次订阅都导致它把数据流从新发一遍

    @Test
    public void testColdPublisher() {
        Flux<String> source = Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
                .map(String::toUpperCase);

        source.subscribe(d -> System.out.println("Subscriber 1: "+d));
        System.out.println();
        source.subscribe(d -> System.out.println("Subscriber 2: "+d));
    }
Hot

手动触发数据流

    @Test
    public void testConnectableFlux() throws InterruptedException {
        Flux<Integer> source = Flux.range(1, 3)
                .doOnSubscribe(s -> System.out.println("上游收到订阅"));

        ConnectableFlux<Integer> co = source.publish();

        co.subscribe(System.out::println, e -> {}, () -> {});
        co.subscribe(System.out::println, e -> {}, () -> {});

        System.out.println("订阅者完成订阅操作");
        Thread.sleep(500);
        System.out.println("还没有连接上");

        co.connect();
    }

调度器和线程模型

前面介绍了响应式流和在其上可以进行的各种操作，Scheduler可以指定这些操作执行的方式和所在的线程。

调度器
  • publishOn调整之后的操作的运行线程
  • subscribeOn设置数据源的操作的运行线程
    @Test
    void testSubscribeOn() {
        Flux<Integer> fluxMap = Flux.range(1, 4)
                .map(integer -> {
                    System.out.println("Map1 number : " + integer +
                            " Thread is : " + Thread.currentThread().getName());
                    return integer;
                })
                .subscribeOn(Schedulers.single())
                .map(integer -> {
                    System.out.println("Map2 number : " + integer +
                            " Thread is : " + Thread.currentThread().getName());
                    return integer;
                });

        StepVerifier.create(fluxMap)
                .expectNext(1, 2, 3, 4)
                .verifyComplete();
    }

    @Test
    void testPublishOn() {
        Flux<Integer> fluxMap = Flux.range(1, 4)
                .map(integer -> {
                    System.out.println("Map1 number : " + integer +
                            " Thread is : " + Thread.currentThread().getName());
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    return integer;
                })
                .publishOn(Schedulers.single())
                .map(integer -> {
                    System.out.println("Map2 number : " + integer +
                            " Thread is : " + Thread.currentThread().getName());
                    return integer;
                });

        StepVerifier.create(fluxMap)
                .expectNext(1, 2, 3, 4)
                .verifyComplete();
    }
线程模型
  • Schedulers.immediate() 当前线程
  • Schedulers.single() 单一的可复用的线程
  • Schedulers.elastic() 弹性大小缓存线程池，线程池中的线程是可以复用的。当所需要时，新的线程会被创建。如果一个线程闲置太长时间，则会被销毁。该调度器适用于 I/O 操作相关的流的处理。
  • Schedulers.parallel() 并行操作优化的线程池，通过 Schedulers.parallel()方法来创建。其中的线程数量取决于 CPU 的核的数量。该调度器适用于计算密集型的流的处理。
    @Test
    // 执行时间大概8秒，创建新线程
    void testNewSingle() {
        System.out.println(LocalDateTime.now());
        Flux.just("tom", "jack", "allen")
                .publishOn(Schedulers.newSingle("1")).map(this::doSomething)
                .publishOn(Schedulers.newSingle("2")).map(this::doSomething)
                .publishOn(Schedulers.newSingle("3")).map(this::doSomething)
                .publishOn(Schedulers.newSingle("4")).map(this::doSomething)
                .publishOn(Schedulers.newSingle("5")).map(this::doSomething)
                .subscribeOn(Schedulers.newSingle("0")).blockLast();
        System.out.println(LocalDateTime.now());
    }

    @Test
    // 执行时间15秒，使用相同线程
    void testSingle() {
        System.out.println(LocalDateTime.now());
        Flux.just("tom", "jack", "allen")
                .publishOn(Schedulers.single()).map(this::doSomething)
                .publishOn(Schedulers.single()).map(this::doSomething)
                .publishOn(Schedulers.single()).map(this::doSomething)
                .publishOn(Schedulers.single()).map(this::doSomething)
                .publishOn(Schedulers.single()).map(this::doSomething)
                .subscribeOn(Schedulers.single()).blockLast();
        System.out.println(LocalDateTime.now());

    }

    private String doSomething(String s) {
        System.out.println(Thread.currentThread().getName());
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return s;
    }

Spring WebFlux

以Reactor框架为基础的，响应式Web框架
传统的Spring MVC基于Servlet，是阻塞式的，每次请求由一个线程处理；而Spring WebFlux通过事件循环Event Loop的方式，由单个线程非阻塞的处理事件。当需要处理耗时任务时，Event Loop绑定的线程会新启线程来执行，完成后通知Event Loop。

Screen Shot 2020-09-23 at 11.46.52 PM.png

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