springboot2 webfux 响应式编程(Mono/Flux/Backpressure)

素颜马尾好姑娘i 2021-11-16 18:26 932阅读 0赞

## What’s Spring WebFlux ##

让我们来看下官方英文怎么介绍的，有兴趣的也可以看一下spring官方的[web-react][]。

> The original web framework included in the Spring Framework, Spring Web MVC, was purpose-built for the Servlet API and Servlet containers. The reactive-stack web framework, Spring WebFlux, was added later in version 5.0. It is fully `non-blocking`, supports Reactive Streams back pressure, and runs on such servers as Netty, Undertow, and Servlet 3.1+ containers.

> Both web frameworks mirror the names of their source modules (spring-webmvc and spring-webflux) and co-exist side by side in the Spring Framework. Each module is optional. Applications can use one or the other module or, in some cases, both — for example, Spring MVC controllers with the reactive WebClient.

`Spring MVC`是Spring Framework中包含的`原始Web框架`，也是专为Servlet API和Servlet容器构建的Web框架。

`响应式堆栈Web框架`是Spring WebFlux是在Spring5版才有的， 是一个完全是`非阻塞`编程，支持Reactive Streams响应流，并且可以很好运行在Netty，Undertow和Servlet 3.1+容器等服务器上（言外之意，开发人员可以使用 WebFlux 创建`高性能` 的 Web 应用和客户端）。

这两个Web框架都反映了其源模块的名称（spring-webmvc和spring-webflux），并在`Spring Framework`中`并存`。 每个模块都是可选的。 应用程序可以使用一个或另一个模块，或者在某些情况下，两者都混合使用，例如在MVC的Controller中带有反应式WebClient。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]  
接下来，我们一起搞清楚几个问题:

1.  **关于Reactive Streams、Srping Reactor 和 Spring WebFlux之间的关系**

> Reactive Streams 是`规范`，Srping Reactor `实现`了 Reactive Streams。Spring WebFlux 是以 Reactor 为`基础`，实现 Web 领域的反应式编程`框架`。

1.  **反应式编程思想是什么？Backpressure背压又是什么？**

> 反应式编程框架主要采用了观察者模式，而 Spring Reactor 的核心则是对观察者模式的一种延伸。观察者模式的架构中`被观察者(Observable)`和`观察者(Subscriber)`处在不同的线程环境中时，由于者各自的工作量不一样，导致它们产生事件和处理事件的速度不一样，这时就出现了两种情况：
> 
>  *  Controller被观察者处理事件慢一些，Client/Browser观察者处理在等啊等。那么Client/Browser观察者就会等着被观察者Controller响应事件，例如排队卖烧饼，被观察者在卖，观察者在买，没问题吧，典型的。
>  *  Controller被观察者处理事件的速度很快，而Client/Browser观察者处理很慢（客户端忙，服务器想响应给你，你都无法接收）。那就有点问题了，如果不作处理的话，事件会堆积起来，最终挤爆你的内存，导致程序崩溃，生产一堆烧饼等人买？
>  *  这时就出现了Backpressure（`背压`\-----指在异步场景中，被观察者Controller发送事件速度远快于Client/Browser观察者的处理速度的情况下，一种告诉上游的被观察者`降低发送速度的策略`）

1.  **既然Webflux不是基于Servlet，那么Spring Security等基于Servlet的组件可以用吗**

> 如果打算将 Web Flux 运行在 Servlet 容器之上，必须是支持 `Servlet 3.1 以上`，因为才有`非阻塞NIO`的支持，虽然 Web Flux 的 API 在某些地方，确实提供了NIO的选项，若单纯只是试着将基于 Web MVC 的应用程序，改写为套用 Web Flux，并不是十分出色的方案。  
> =。=  
> 例如，Spring Security 显然就`不能用`了，毕竟是 Spring 基于 Servlet 的安全方案，开发者必须想办法改造为 `Spring Security Reactive`；而且，在储存方案上，也不是直接采用 Spring Data，而是建议 Spring Data Reactive 等，因为现在整个生态都不完善，所以暂时不会有很多东西出来，但是肯定会慢慢完善。

1.  **如何更好的理解Mono和Flux**

>  *  Mono / Flux 都是`Publisher（发布者）`，也就是`被观察者(Observable)`。
>  *  Mono / Flux 都实现了 `org.reactivestreams.Publisher<T>` 接口，在 Web Flux，你的方法只需返回 Mono / Flux 即可。你的代码基本也只和 Mono 或 Flux 打交道。而 Web Flux 则会实现 Subscriber ，调用onNext()时将业务开发人员编写的 Mono/Flux 转换为 HTTP Response 返回给客户端。

## Maven依赖 ##

<dependency>
    	<groupId>org.springframework.boot</groupId>
    	<artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>

## Entity ##

简单构造一个请求和响应的业务实体

@Data
    public class User { 
    	private String id;
    	private String name;
    	private String mobile;
    	private Integer age;
    }

## Controller ##

WebFlux复杂？一个controller搞定hello world。

import java.util.Map;
    import java.util.Objects;
    import java.util.concurrent.ConcurrentHashMap;
    
    import javax.management.relation.RelationNotFoundException;
    
    import org.springframework.http.HttpStatus;
    import org.springframework.web.bind.annotation.DeleteMapping;
    import org.springframework.web.bind.annotation.ExceptionHandler;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.PostMapping;
    import org.springframework.web.bind.annotation.PutMapping;
    import org.springframework.web.bind.annotation.RequestBody;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseStatus;
    import org.springframework.web.bind.annotation.RestController;
    
    import com.softdev.system.demo.entity.User;
    
    import reactor.core.publisher.Flux;
    import reactor.core.publisher.Mono;
    
    @RestController
    @RequestMapping("/user")
    public class UserController { 
    	
        private final Map<String, User> data = new ConcurrentHashMap<>();
        
        @ResponseStatus(value = HttpStatus.NOT_FOUND, reason = "Resource not found")
        @ExceptionHandler(RelationNotFoundException.class)
        public void notFound() { 
        }
     
        @GetMapping("")
        public Flux<User> list() { 
        	try { 
    			Thread.sleep(5000);
    		} catch (InterruptedException e) { 
    		}
            return Flux.fromIterable(this.data.values());
        }
     
        @GetMapping("/{id}")
        public Mono<User>getById(@PathVariable("id") final String id) { 
        	return Mono.justOrEmpty(this.data.get(id))
                    .switchIfEmpty(Mono.error(new RelationNotFoundException("ResourceNotFoundException")));
        }
     
        @PostMapping("")
        public Mono<User> create(@RequestBody final User user) { 
        	this.data.put(user.getId(), user);
            return Mono.just(user);
        }
     
        @PutMapping("/{id}")
        public Mono<User>  update(@PathVariable("id") final String id, @RequestBody final User user) { 
            Objects.requireNonNull(user);
            user.setId(id);
            this.data.put(user.getId(), user);
            return Mono.just(user);
        }
     
        @DeleteMapping("/{id}")
        public Mono<User>  delete(@PathVariable("id") final String id) { 
        	return Mono.justOrEmpty(this.data.remove(id));
        }
    }

## Postman并发压测 ##

> 1.  **创建用户**  
>     curl -X POST [http://localhost:9999/webflux/user][http_localhost_9999_webflux_user]  
>     \-d ‘\{“id”:“10001”,“name”:“[zhengkai.blog.csdn.net][]”,“mobile”:“18522222222”,“age”:18\}’  
>     ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]  
>     2.**创建Collection和Request**  
>     ![在这里插入图片描述][20190729113824947.png]  
>     curl -X GET [http://localhost:9999/webflux/user][http_localhost_9999_webflux_user]
> 2.  **运行Runner**  
>     ![>][20190729114610681.png]  
>     运行单个是单线程测试，运行多个可以进行多线程测试  
>     单线程测试：5ms阻塞的响应  
>     ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 2]  
>     同时运行两个，双线程测试（=。=postman为何不直接提供多线程测试）：  
>     `无阻塞模式。都是5ms响应两条线程`  
>     ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 3]

## Reactive化改造? ##

目前官方`reactive`化的资料，隐约feel到一个场景，就是用weblux+redis做热点数据的读取，应该超级无敌爽，比servlet+cache快。当然现在只是有很多的资料，用的人多了，晚点应该就有stater了：

> [https://spring.io/guides/gs/spring-data-reactive-redis/][https_spring.io_guides_gs_spring-data-reactive-redis]  
> [https://spring.io/blog/2016/11/28/going-reactive-with-spring-data][https_spring.io_blog_2016_11_28_going-reactive-with-spring-data]  
> [https://spring.io/blog/2017/10/04/spring-tips-reactive-spring-security][https_spring.io_blog_2017_10_04_spring-tips-reactive-spring-security]  
> [https://spring.io/blog/2019/05/16/reactive-transactions-with-spring][https_spring.io_blog_2019_05_16_reactive-transactions-with-spring]

## How to control backpressure? ##

如果需要对backpressure背压进行控制，我们可以使用Reactor Operators进行控制，也就是 `limitRate()` 方法，very convince。

@PostMapping("/tweets")
    public Mono<Void> postAllTweets(Flux<Tweet> tweetsFlux) { 
        return tweetService.process(tweetsFlux.limitRate(10)).then();
    }
    @GetMapping("/tweets")
    public Flux<Tweet> getAllTweets() { 
        return tweetService.retreiveAll().limitRate(10);
    }

[web-react]: https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html#webflux
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70]: /images/20211116/e7575a6a951d47d9b5711ee06eba0f8d.png
[http_localhost_9999_webflux_user]: http://localhost:9999/webflux/user
[zhengkai.blog.csdn.net]: http://zhengkai.blog.csdn.net
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 1]: /images/20211116/66ec1879d7d6421e8dcba5737c1d8f53.png
[20190729113824947.png]: /images/20211116/22e3aed1119f4a64903be5dc28982456.png
[20190729114610681.png]: /images/20211116/1bc53128d86145d09d272a713c2d02ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 2]: /images/20211116/d93786d3e21241c89d888083a0cda432.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly96aGVuZ2thaS5ibG9nLmNzZG4ubmV0_size_16_color_FFFFFF_t_70 3]: /images/20211116/96d660ebf1a04ea1bd034189e2203ccb.png
[https_spring.io_guides_gs_spring-data-reactive-redis]: https://spring.io/guides/gs/spring-data-reactive-redis/
[https_spring.io_blog_2016_11_28_going-reactive-with-spring-data]: https://spring.io/blog/2016/11/28/going-reactive-with-spring-data
[https_spring.io_blog_2017_10_04_spring-tips-reactive-spring-security]: https://spring.io/blog/2017/10/04/spring-tips-reactive-spring-security
[https_spring.io_blog_2019_05_16_reactive-transactions-with-spring]: https://spring.io/blog/2019/05/16/reactive-transactions-with-spring