参考：
https://reflectoring.io/reactive-architecture-with-spring-boot/
- Reactive System
首先介绍下Reactive System。
通常情况下，在服务间数据的传输是阻塞的、同步的、先进先出，阻塞式的传输会影响效率。在这基础上，开发人员发起了基于reactive的系统架构，具体的reactive系统定义可以查看：
2. Spring Reactive
在Spring官网https://spring.io/ 首页，可以看到Reactive项目排名是非常靠前的，仅次于大热的微服务，可见Spring是很重视这个项目的：
在Spring Reactive首页：https://spring.io/reactive，拉到底下有张图，能很好的描述Reactive系统的架构：
具体来说：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个接口：
使用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个。
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的例子：其中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
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中的数据为空：
打开swagger-ui：http://localhost:9000/swagger-ui.html
试图插入user.id="1"的数据：
也可通过单元测试来测试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()
