（七）springboot实战——springboot3集成R2DBC实现webflux响应式编程服务案例

前言

本节主要内容是关于使用新版springboot3集成响应式数据库R2DBC,完成响应式web服务案例。需要注意的是，此次项目使用的JDK版本是JDK17，springboot版本使用3.2.2版本，数据库使用关系型数据库mysql。WebFlux 是一个基于响应式编程模型的框架，适用于构建异步、非阻塞的高性能 Web 应用程序。它具有高并发能力、函数式编程风格、与其他 Spring 框架的集成能力等优势，可以满足对性能要求高或需要处理大量并发请求的场景。

WebFlux 使用响应式编程模型，基于 Reactor 库实现。它利用异步和非阻塞的方式处理请求，能够处理大量的并发请求，提高应用程序的性能和可扩展性。由于采用了非阻塞的 I/O 操作，WebFlux 能够更有效地利用服务器资源，处理更多的并发请求，提供更高的吞吐量。WebFlux 鼓励使用函数式和声明式的编程风格，通过组合一系列操作来处理请求和响应。这种编程方式更具可读性和可维护性。WebFlux 使用 Flux 和 Mono 这两个反应式类型来处理异步数据流。Flux 用于表示包含零个或多个元素的异步序列，而 Mono 用于表示包含零个或一个元素的异步结果。WebFlux 可以与其他 Spring 框架和库进行无缝集成，例如 Spring Boot、Spring Data、Spring Security 等。这使得开发者可以充分利用 Spring 生态系统的优势。WebFlux 不依赖于特定的服务器，可以在各种服务器上运行，包括 Netty、Undertow 和 Tomcat 等。这为开发者提供了更多的灵活性和选择性。

正文

①使用idea创建一个springboot3项目

②引入r2dbc的mysql驱动和响应式Spring Data R2dbc启动器以及接口文档springdoc

<dependency>
     <groupId>org.projectlombok</groupId>
     <artifactId>lombok</artifactId>
</dependency>

<!-- https://mvnrepository.com/artifact/org.springdoc/springdoc-openapi-starter-webflux-ui -->
<dependency>
	<groupId>org.springdoc</groupId>
	<artifactId>springdoc-openapi-starter-webflux-ui</artifactId>
	<version>2.3.0</version>
</dependency>

<!--        响应式 Spring Data R2dbc-->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-r2dbc</artifactId>
</dependency>

<!-- https://mvnrepository.com/artifact/io.asyncer/r2dbc-mysql -->
<dependency>
	<groupId>io.asyncer</groupId>
	<artifactId>r2dbc-mysql</artifactId>
	<version>1.0.6</version>
</dependency>

③在resource目录下创建配置文件application.yaml，配置mysql的连接配置信息和接口文档

spring:
  r2dbc:
    url: r2dbc:mysql://192.168.110.88:3306/ht-atp
    username: root
    password: root
server:
  port: 8080
springdoc:
  api-docs:
    enabled: true
    path: /api-docs
  swagger-ui:
    enabled: true
    path: /swagger-ui/index.html
  show-actuator: true

④启动项目，验证项目是否可以正常启动，以及接口文档是否可以正常使用

⑤创建用户实体类User

package com.yundi.atp.entity;

import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;


@Schema(name = "User", description = "用户类")
@Table(name = "user")
@Data
public class User {
    @Id
    @Schema(name = "id", description = "用户ID")
    private Integer id;

    @Schema(name = "name", description = "用户名称")
    private String name;

    @Schema(name = "age", description = "用户年龄")
    private Integer age;
}

⑥创建统一接口响应对象ApiResponse

package com.yundi.atp.common;

import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Builder;
import lombok.Data;


@Data
@Builder
public class ApiResponse<T> {
    @Schema(name = "code", description = "状态码")
    private Integer code;

    @Schema(name = "msg", description = "消息结果")
    private String msg;

    @Schema(name = "data", description = "数据")
    private T data;

    /**
     * 成功统一响应格式
     *
     * @param
     * @param <T>
     * @return
     */
    public static <T> ApiResponse<T> success() {
        ApiResponse<T> apiResponse = new ApiResponse<>(200, "成功", null);
        return apiResponse;
    }

    /**
     * 成功统一响应格式
     *
     * @param data
     * @param <T>
     * @return
     */
    public static <T> ApiResponse<T> success(T data) {
        ApiResponse<T> apiResponse = new ApiResponse<>(200, "成功", data);
        return apiResponse;
    }

    /**
     * 失败统一响应格式
     *
     * @param code
     * @param message
     * @param <T>
     * @return
     */
    public static <T> ApiResponse<T> fail(Integer code, String message) {
        return new ApiResponse<>(code, message, null);
    }
}

⑦创建UserController控制类，完成用户管理接口申明

package com.yundi.atp.controller;

import com.yundi.atp.common.ApiResponse;
import com.yundi.atp.entity.User;
import com.yundi.atp.service.UserService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;


@Tag(name = "用户管理", description = "用户管理")
@RestController
@RequestMapping("/user")
public class UserController {
    @Autowired
    private UserService userService;

    @Operation(summary = "获取全部用户信息", description = "获取全部用户信息")
    @GetMapping(value = "getAllUserInfo")
    public Mono<ApiResponse<List<User>>> getAllUserInfo() {
        Flux<User> userInfo = userService.getAllUserInfo();
        return userInfo.collectList().map(user -> ApiResponse.success(user));
    }

    @Operation(summary = "根据ID获取用户信息", description = "根据ID获取用户信息")
    @GetMapping(value = "getUserById/{id}")
    public Mono<ApiResponse<User>> getUserById(@PathVariable("id") Integer id) {
        Mono<User> user = userService.getUserById(id);
        Mono<ApiResponse<User>> userMono = user.map(ApiResponse::success);
        return userMono;
    }

    @Operation(summary = "新增用户", description = "新增用户")
    @PostMapping(value = "saveUser")
    public Mono<ApiResponse> saveUser(@RequestBody User user) {
        Mono<User> userMono = userService.saveUser(user);
        return userMono.thenReturn(ApiResponse.success());
    }

    @Operation(summary = "更新用户", description = "更新用户")
    @PostMapping(value = "updateUser")
    public Mono<ApiResponse> updateUser(@RequestBody User user) {
        Mono<User> userMono = userService.updateUser(user);
        return userMono.thenReturn(ApiResponse.success());
    }

    @Operation(summary = "删除用户", description = "删除用户")
    @DeleteMapping(value = "deleteUser/{id}")
    public Mono<ApiResponse> deleteUser(@PathVariable("id") Integer id) {
        Mono<Void> userMono = userService.deleteUser(id);
        return userMono.thenReturn(ApiResponse.success());
    }
}

⑧ 创建用户管理接口层UserService，定义接口

package com.yundi.atp.service;

import com.yundi.atp.entity.User;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public interface UserService {
    /**
     * 查询全部用户信息
     *
     * @return
     */
    Flux<User> getAllUserInfo();

    /**
     * 根据ID获取用户信息
     *
     * @param id
     * @return
     */
    Mono<User> getUserById(Integer id);

    /**
     * 新增用户
     *
     * @param user
     * @return
     */
    Mono<User> saveUser(User user);

    /**
     * 更新用户
     *
     * @param user
     * @return
     */
    Mono<User> updateUser(User user);

    /**
     * 删除用户
     *
     * @param id
     * @return
     */
    Mono<Void> deleteUser(Integer id);
}

⑨创建用户管理实现层UserServiceImpl，完成用户管理的业务逻辑

package com.yundi.atp.service.impl;

import com.yundi.atp.entity.User;
import com.yundi.atp.repository.UserRepository;
import com.yundi.atp.service.UserService;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class UserServiceImpl implements UserService {
    @Resource
    private UserRepository userRepository;

    @Override
    public Flux<User> getAllUserInfo() {
        return userRepository.findAll();
    }

    @Override
    public Mono<User> getUserById(Integer id) {
        return userRepository.findById(id);
    }

    @Override
    public Mono<User> saveUser(User user) {
        return userRepository.save(user);
    }

    @Override
    public Mono<User> updateUser(User user) {
        return userRepository.updateUser(user);
    }

    @Override
    public Mono<Void> deleteUser(Integer id) {
        return userRepository.deleteById(id);
    }
}

⑩ 创建UserRepository持久层，并继承R2dbcRepository通用数据库操作接口完成数据库的操作

package com.yundi.atp.repository;

import com.yundi.atp.entity.User;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.r2dbc.repository.R2dbcRepository;
import org.springframework.data.repository.query.Param;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Mono;


@Repository
public interface UserRepository extends R2dbcRepository<User, Integer> {
    /**
     * 更新用户信息
     *
     * @param user
     * @return
     */
    @Query("update user set name = :#{#user.name}, age = :#{#user.age} where id = :#{#user.id}")
    Mono<User> updateUser(@Param("user") User user);


    /**
     * 动态更新用户信息
     *
     * @param user
     * @return
     */
    @Query("UPDATE User SET " +
            "name = CASE WHEN :#{#user.name} IS NULL THEN name ELSE :#{#user.name} END, " +
            "age = CASE WHEN :#{#user.age} IS NULL THEN age ELSE :#{#user.age} END " +
            "WHERE id = :#{#user.id}")
    Mono<User> updateUserQuery(@Param("user") User user);
}