【Spring Reactive】Spring WebFlux入

参考：
https://reflectoring.io/reactive-architecture-with-spring-boot/

1. Reactive System
   首先介绍下Reactive System。
   通常情况下，在服务间数据的传输是阻塞的、同步的、先进先出，阻塞式的传输会影响效率。在这基础上，开发人员发起了基于reactive的系统架构，具体的reactive系统定义可以查看：

• 英文版：https://www.reactivemanifesto.org/
• 中文版：https://www.reactivemanifesto.org/zh-CN

2. Spring Reactive

在Spring官网https://spring.io/ 首页，可以看到Reactive项目排名是非常靠前的，仅次于大热的微服务，可见Spring是很重视这个项目的：

image.png

在Spring Reactive首页：https://spring.io/reactive，拉到底下有张图，能很好的描述Reactive系统的架构：

image.png

具体来说：Spring Reactive技术栈包含了一系列的技术，上述列举了各个技术层（如sucurity方面，web方面或是持久层方面）和现有的框架做了对比。其中最重要的是WebFlux框架，即反应式web框架，对标Spring MVC，是Reactive微服务集成Spring Boot2中重要的一环。在介绍Spring WebFlux之前，先介绍下Reactive相关的技术。

关于Spring Reactive：

• 官方例子：https://spring.io/guides/gs/reactive-rest-service/
• Spring Reactive文档：https://docs.spring.io/spring-framework/docs/current/reference/html/web-reactive.html

3. Reactive相关的技术

3.1 首先是Reactive Stream

具体看：https://www.reactive-streams.org/

JDK9提供了Reactive Streams的简单实现，主要是基于类Flow，java doc：https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Flow.html
主要是提供了4个接口：

image.png

使用Flow下面4个接口的demo演示，可以参考网友的博文：https://blog.csdn.net/qq_34285557/article/details/104569122

Publisher可以发送元素，Subscriber接收元素并做出响应。当执行Publisher里的subscribe方法时，发布者会回调订阅者的onSubscribe方法，这个方法中，通常订阅者会借助传入的Subscription向发布者请求n个数据。然后发布者通过不断调用订阅者的onNext方法向订阅者发出最多n个数据。如果数据全部发完，则会调用onComplete告知订阅者流已经发完；如果有错误发生，则通过onError发出错误数据，同样也会终止流。

默认情况下，如果subscriber方拉取数据比较慢，那么默认的缓冲带大小为256个。 image.png 比如我们在subscriber的onNext(item)方法上加上Thread.sleep，那么生产数据就会在第258个处等待，原因是生产的第一个数据则会在onSubscribe的时候拿出，然后缓冲区为256个，那么接下来的则需要等Thread.sleep时间到后才会被拿走： image.png

3.2 Project Reactor

• 官网：https://projectreactor.io/
• 最新Release文档：https://projectreactor.io/docs/core/3.5.2/reference/index.html
• baeldung网站上的hello world例子：https://www.baeldung.com/reactor-core

基于上述的Reactive Stream，可与Java的functional API, CompletableFuture, Stream以及Duration结合使用。Spring很多项目中都用到了该库，如Spring WebFlux, Spring Data以及Spring Cloud Gateway。

Reactor提供了两类Publisher：

• Flux[N]：可异步发布0-N个元素，官方doc。
• Mono[0|1]：可异步发布0或1个元素，官方doc。

Flux的例子：其中logs()是为了打印日志：

        Flux.range(1,3).log().subscribe(s -> log.info("received: {}", s));

运行后打印，可以看到和Reactive Stream一样，也有个subscribe的过程，request以及onNext的过程：

14:39:04.339 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
14:39:04.351 [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription)
14:39:04.353 [main] INFO reactor.Flux.Range.1 - | request(unbounded)
14:39:04.354 [main] INFO reactor.Flux.Range.1 - | onNext(1)
14:39:04.354 [main] INFO com.reactor.FluxDemo - received: 1
14:39:04.354 [main] INFO reactor.Flux.Range.1 - | onNext(2)
14:39:04.354 [main] INFO com.reactor.FluxDemo - received: 2
14:39:04.354 [main] INFO reactor.Flux.Range.1 - | onNext(3)
14:39:04.354 [main] INFO com.reactor.FluxDemo - received: 3
14:39:04.354 [main] INFO reactor.Flux.Range.1 - | onComplete()

Mono的例子：

        Mono.just("John").log().subscribe(System.out::println);

运行后日志：

15:36:56.824 [main] DEBUG reactor.util.Loggers - Using Slf4j logging framework
15:36:56.844 [main] INFO reactor.Mono.Just.1 - | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
15:36:56.847 [main] INFO reactor.Mono.Just.1 - | request(unbounded)
15:36:56.848 [main] INFO reactor.Mono.Just.1 - | onNext(John)
John
15:36:56.849 [main] INFO reactor.Mono.Just.1 - | onComplete()

4. Spring WebFlux

从Spring 5开始，Spring增加了reactive的web框架：Spring WebFlux。从官方文档上看，Spring WebMvc和Spring WebFlux可以单独分开使用，也可一起使用。通过Spring WebFlux可以构建异步式的web项目，使用reactive stream以及functional的API来来更好的支持并发。

Spring WebFlux官方文档：https://docs.spring.io/spring-framework/docs/current/reference/html/web-reactive.html#webflux

Spring WebFlux支持两种编程模式，所谓的两种编程模式是在Controller这层的变化：

• Annotated Controllers：和Spring MVC的编程风格保持一致，也可使用类似@GetMapping，@PostMapping这样的注解。
• Functional Endpoints: lambda风格的，使用functional的编程方式。可以看做是在app中有一系列的工具类，我们需要给它们加上路由以使可以支持requests。

4.1 Spring WebFlux demo

【baeldung网站上的hello world的例子】

• 基地Annotated Controllers的例子：https://www.baeldung.com/spring-webflux
• 基于Functional Endpoints的例子：https://www.baeldung.com/spring-5-functional-web

【此外还有一些学习的例子也很不错】

• https://reflectoring.io/reactive-architecture-with-spring-boot/
• Getting Started with Spring WebFlux --> 【本文参照的是这篇文章搭建的demo。】

4.2 安装mongodb

持久层我们选择mongodb，所以需要先安装mongodb，使用的是docker安装。

• 参考：https://www.linode.com/docs/guides/set-up-mongodb-on-docker/
• 官网镜像地址：https://hub.docker.com/_/mongo

拉取镜像：

docker pull mongo

启动：

docker run --name mongo_example -d -p 27017:27017 --name mongo_example -e MONGO_INITDB_ROOT_USERNAME=mongoadmin -e MONGO_INITDB_ROOT_PASSWORD=mypass mongo

怎样进入mongo bash：

docker exec -it mongo_example bash

进入后再次输入，需要按提示输入密码：mypass：

mongosh -u mongoadmin

image.png

4.3 引入依赖

• spring-boot-starter-webflux是Spring WebFlux的主要依赖
• spring-boot-starter-data-mongodb-reactive是持久化层mongodb基于reactive实现的依赖
• springdoc-openapi-webflux-ui是Spring Doc基于webflux的依赖，主要是为了swagger-ui页面，可以更好的测试代码。

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springdoc</groupId>
            <artifactId>springdoc-openapi-webflux-ui</artifactId>
            <version>1.6.4</version>
        </dependency>

4.4 application.yaml配置

首先是mongodb的配置，这里的username/password需要跟上面docker安装的一致。
其次是Netty相关的配置，使用的是9000端口：

spring:
  data:
    mongodb:
      host: 127.0.0.1
      port: 27017
      username: mongoadmin
      password: mypass
      database: test
      authenticationDatabase: admin

server:
  port: 9000
  http2:
    enabled: true

4.5 首先是POJO层

新建User类，其中@Document(value = "users")是声明使用的是Mongodb：

@ToString
@EqualsAndHashCode(of = {"id","name","department"})
@AllArgsConstructor
@NoArgsConstructor
@Data
@Document(value = "users")
public class User {

    @Id
    private String id;
    private String name;
    private int age;
    private double salary;
    private String department;
}

Repository：

@Repository
public interface UserRepository extends ReactiveMongoRepository<User, String> {
}

4.6 新建Service层：

@Service
@Slf4j
@RequiredArgsConstructor
@Transactional
public class UserService {

    private final ReactiveMongoTemplate reactiveMongoTemplate;
    private final UserRepository userRepository;

    public Mono<User> createUser(User user){
        return userRepository.save(user);
    }

    public Flux<User> getAllUsers(){
        return userRepository.findAll();
    }

    public Mono<User> findById(String userId){
        return userRepository.findById(userId);
    }

    public Mono<User> updateUser(String userId,  User user){
        return userRepository.findById(userId)
                .flatMap(dbUser -> {
                    dbUser.setAge(user.getAge());
                    dbUser.setSalary(user.getSalary());
                    return userRepository.save(dbUser);
                });
    }

    public Mono<User> deleteUser(String userId){
        return userRepository.findById(userId)
                .flatMap(existingUser -> userRepository.delete(existingUser)
                        .then(Mono.just(existingUser)));
    }

    public Flux<User> fetchUsers(String name) {
        Query query = new Query().with(Sort.by(Collections.singletonList(Sort.Order.asc("age"))));
        query.addCriteria(Criteria.where("name").regex(name));
        return reactiveMongoTemplate.find(query, User.class);
    }
}

4.7 Controller层

@RequiredArgsConstructor
@RestController
@RequestMapping("/users")
public class UserController {

    private final UserService userService;

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<User> create(@RequestBody User user){
        return userService.createUser(user);
    }

    @GetMapping
    public Flux<User> getAllUsers(){
        return userService.getAllUsers();
    }

    @GetMapping("/{userId}")
    public Mono<ResponseEntity<User>> getUserById(@PathVariable String userId){
        Mono<User> user = userService.findById(userId);
        return user.map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    @PutMapping("/{userId}")
    public Mono<ResponseEntity<User>> updateUserById(@PathVariable String userId, @RequestBody User user){
        return userService.updateUser(userId,user)
                .map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.badRequest().build());
    }

    @DeleteMapping("/{userId}")
    public Mono<ResponseEntity<Void>> deleteUserById(@PathVariable String userId){
        return userService.deleteUser(userId)
                .map( r -> ResponseEntity.ok().<Void>build())
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    @GetMapping("/search")
    public Flux<User> searchUsers(@RequestParam("name") String name) {
        return userService.fetchUsers(name);
    }

    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<User> streamAllUsers() {
        return userService
                .getAllUsers()
                .flatMap(user -> Flux.zip(Flux.interval(Duration.ofSeconds(2)),
                                Flux.fromStream(Stream.generate(() -> user))).map(Tuple2::getT2));
    }
}

4.8 新建Spring Boot启动类

@EnableMongoAuditing
@EnableReactiveMongoRepositories
@SpringBootApplication
public class WebFluxApplication {
    public static void main(String[] args) {
        SpringApplication.run(WebFluxApplication.class, args);
    }
}

4.9 测试

在没有插入任何数据前，mongodb中的数据为空： image.png

打开swagger-ui：http://localhost:9000/swagger-ui.html

试图插入user.id="1"的数据： image.png 插入成功： image.png 查询mongodb数据库，插入成功： image.png 也可通过swagger-ui查询： image.png

也可通过单元测试来测试UserRepository：
参考：https://github.com/eugenp/tutorials/tree/master/persistence-modules/spring-data-mongodb-reactive/src/test/java/com/baeldung/reactive/repository

@SpringBootTest
public class UserRepositoryTest {
    @Autowired
    private UserRepository userRepository;

    @Test
    public void tes() {
        Mono<User> userMono = userRepository.findById("1");
        StepVerifier.create(userMono)
                .assertNext(u -> {
                    Assertions.assertEquals("bill", u.getName());
                })
                .expectComplete()
                .verify();
    }
}

也可以用WebClient测试：
参考：https://github.com/eugenp/tutorials/blob/master/spring-reactive-modules/spring-reactive/src/main/java/com/baeldung/reactive/webclient/WebController.java

@Slf4j
@SpringBootTest(classes = WebFluxApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class EmployeeWebClientTest {

    @LocalServerPort
    private int port;
    
    @Test
    public void whenBindToController_thenRequestProcessed() throws InterruptedException {
        Flux<User> userFlux = WebClient.create()
                .get()
                .uri("http://localhost:" + port + "/users")
                .retrieve()
                .bodyToFlux(User.class);

        userFlux.log().subscribe(user -> log.info(user.toString()));

        Thread.sleep(100000L);
    }
}

日志：

2023-01-26 23:20:53.789 INFO 38308 --- [ main] reactor.Flux.MonoFlatMapMany.1 : onSubscribe(MonoFlatMapMany.FlatMapManyMain)
2023-01-26 23:20:53.791 INFO 38308 --- [ main] reactor.Flux.MonoFlatMapMany.1 : request(unbounded)
2023-01-26 23:20:55.051 INFO 38308 --- [ntLoopGroup-3-3] org.mongodb.driver.connection : Opened connection [connectionId{localValue:3, serverValue:112}] to 127.0.0.1:27017
2023-01-26 23:20:55.175 INFO 38308 --- [ctor-http-nio-3] reactor.Flux.MonoFlatMapMany.1 : onNext(User(id=1, name=bill, age=18, salary=1000.0, department=dept-a))
2023-01-26 23:20:55.182 INFO 38308 --- [ctor-http-nio-3] com.webflux.EmployeeWebClientTest : User(id=1, name=bill, age=18, salary=1000.0, department=dept-a)
2023-01-26 23:20:55.182 INFO 38308 --- [ctor-http-nio-3] reactor.Flux.MonoFlatMapMany.1 : onComplete()