4.2 安裝mongodb

持久層我們選擇mongodb，所以需要先安裝mongodb，使用的是docker安裝。

• 參考：https://www.linode.com/docs/guides/set-up-mongodb-on-docker/
• 官網(wǎng)鏡像地址：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實現(xiàn)的依賴
• 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相關(guān)的配置，使用的是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 測試

image.png

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

image.png

image.png

image.png

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