（七）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-atpusername: rootpassword: root
server:port: 8080
springdoc:api-docs:enabled: truepath: /api-docsswagger-ui:enabled: truepath: /swagger-ui/index.htmlshow-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 {@Autowiredprivate 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 {@Resourceprivate UserRepository userRepository;@Overridepublic Flux<User> getAllUserInfo() {return userRepository.findAll();}@Overridepublic Mono<User> getUserById(Integer id) {return userRepository.findById(id);}@Overridepublic Mono<User> saveUser(User user) {return userRepository.save(user);}@Overridepublic Mono<User> updateUser(User user) {return userRepository.updateUser(user);}@Overridepublic 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);
}