美文网首页
【Spring Reactive】Spring WebFlux入

【Spring Reactive】Spring WebFlux入

作者: 伊丽莎白2015 | 来源:发表于2023-01-25 23:22 被阅读0次

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

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

    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：

    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

    基于上述的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的例子】

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

    4.2 安装mongodb

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

    拉取镜像：

    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()

    相关文章

      网友评论

          本文标题：【Spring Reactive】Spring WebFlux入

          本文链接：https://www.haomeiwen.com/subject/tltdhdtx.html

          延伸阅读

          深度阅读

          • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

          栏目导航

          热点阅读

          关于我们|服务条款|联系我们|【Spring Reactive】Spring WebFlux入|投稿指南|网站地图|RSS订阅|排版工具|手机版

          提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

          Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

          备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

          本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

          所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！