关系数据构建反应式的spring驱动程序 原 荐

先说说什么是响应式

响应式编程或反应式编程（英语：Reactive programming）是一种面向数据流和变化传播的编程范式，直白的说就是：将变化的值通过数据流进行传播。

反应式架构文章

• WebFlux定点推送、全推送灵活websocket运用【推】
• WebFlux前后端分离 -- 数据响应式展示 【推】
• webflux+redis/mongodb提供响应式API【推】

WebFlux是什么呢

WebFlux 模块的名称是 spring-webflux，名称中的 Flux 来源于 Reactor 中的类 Flux。Spring webflux 有一个全新的非堵塞的函数式 Reactive Web 框架，可以用来构建异步的、非堵塞的、事件驱动的服务，在伸缩性方面表现非常好。

spring-webflux 模块。该模块包含对响应式 HTTP 和 WebSocket 客户端的支持，以及对 REST，HTML 和 WebSocket 交互等程序的支持。一般来说，Spring MVC 用于同步处理，Spring Webflux 用于异步处理。

Spring Boot Webflux 有两种编程模型实现，一种类似 Spring MVC 注解方式，另一种是基于 Reactor 的响应式方式。

R2DBC是什么？

响应式关系数据库连接的支持

主题

之前我有对非关系性数据库的响应式api实现（传送门），而今天我们要说说非关系型数据库实现。

体验spring boot 2.3.0.RC1 , 期待已久mysql  reactive 总算开始展露头角了。 今天就来整合一下springboot 和r2dbc 与mysql。主要是采用 Spring Data R2DBC ，Spring Data R2DBC对R2DBC应用了熟悉的Spring抽象和存储库支持。它使在响应式应用程序堆栈中构建使用关系数据访问技术的Spring驱动的应用程序更加容易。

POM

<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<parent>
		<groupId>com.gitee.occo</groupId>
		<artifactId>occo-parent</artifactId>
		<version>2.0.0-SNAPSHOT</version>
	</parent>
	<artifactId>occo-sso</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<dependencies>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-webflux</artifactId>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-r2dbc</artifactId>
		</dependency>

		<dependency>
			<groupId>dev.miku</groupId>
			<artifactId>r2dbc-mysql</artifactId>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<scope>runtime</scope>
		</dependency>
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
			<exclusions>
				<exclusion>
					<groupId>org.junit.vintage</groupId>
					<artifactId>junit-vintage-engine</artifactId>
				</exclusion>
			</exclusions>
		</dependency>
		<dependency>
			<groupId>io.projectreactor</groupId>
			<artifactId>reactor-test</artifactId>
			<scope>test</scope>
		</dependency>

		<dependency>
			<groupId>org.projectlombok</groupId>
			<artifactId>lombok</artifactId>
			<scope>provided</scope>
		</dependency>
	</dependencies>

	<build>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<configuration>
					<source>${java.version}</source>
					<target>${java.version}</target>
					<encoding>UTF-8</encoding>
				</configuration>
			</plugin>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-release-plugin</artifactId>
				<configuration>
					<tagNameFormat>v@{project.version}</tagNameFormat>
					<autoVersionSubmodules>true</autoVersionSubmodules>
					<indentSize>4</indentSize>
					<useEditMode>true</useEditMode>
					<localCheckout>true</localCheckout>
				</configuration>
			</plugin>
		</plugins>
	</build>
</project>

POM中的occo-parent可以替换成

<parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>2.3.0.RC1</version>
	<relativePath/> <!-- lookup parent from repository -->
</parent>


<!--下不了jar加上下面的：-->
    <repositories>
		<repository>
			<id>spring-milestones</id>
			<name>Spring Milestones</name>
			<url>https://repo.spring.io/milestone</url>
		</repository>
	</repositories>
	<pluginRepositories>
		<pluginRepository>
			<id>spring-milestones</id>
			<name>Spring Milestones</name>
			<url>https://repo.spring.io/milestone</url>
		</pluginRepository>
	</pluginRepositories>

application.yml

server:
  port: 8000
  
spring:
  application:
    name: occo-sso
  r2dbc: 
    url: r2dbc:mysql://localhost:3306/occo
    username: root
    password: pwd2020
logging:
  level:
    org.springframework.data.r2dbc: DEBUG

实现一个简单数据操作

repository：

import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;

import com.gitee.occo.sso.user.entity.User;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Repository
public interface UserRepository extends ReactiveCrudRepository<User, Long>{

	@Query("SELECT * FROM user WHERE state = :state")
	Flux<User> findByState(Integer state);
	
	@Query("SELECT * FROM user WHERE username = :username")
	Mono<User> findByUsername(String username);
	
	@Query("SELECT * FROM user WHERE username LIKE :username")
	Flux<User> findByUsernameLike(String username);
}

service

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.gitee.occo.sso.user.entity.User;
import com.gitee.occo.sso.user.repository.UserRepository;
import com.gitee.occo.sso.user.service.UserService;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class UserServiceImpl implements UserService {

	@Autowired
	private UserRepository userRepository;

	@Override
	public Mono<User> findById(Long id) {
		// TODO Auto-generated method stub
		return userRepository.findById(id);
	}

	@Override
	public Mono<User> findByUsername(String username) {
		// TODO Auto-generated method stub
		return userRepository.findByUsername(username);
	}

	@Override
	public Flux<User> findByUsernameLike(String username) {
		// TODO Auto-generated method stub
		return userRepository.findByUsernameLike(username);
	}

}

controller:

import java.time.Duration;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.gitee.occo.sso.user.entity.User;
import com.gitee.occo.sso.user.service.UserService;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/user")
public class UserController {

	@Autowired
	UserService userService;
	
	@GetMapping("/findById/{id}")
	public Mono<User> getUserById(@PathVariable("id")Long id){
		return userService.findById(id);
	}
	
	@GetMapping("/findByUsername")
	public Mono<User> findByUsername(String username){
		return userService.findByUsername(username);
	}
	
	@GetMapping(value="/findByUsernameLike" ,produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
	public Flux<User> getByUsername(String username){
		return userService.findByUsernameLike(username+"%").delayElements(Duration.ofSeconds(2));
	}
}

PS：produces = MediaType.APPLICATION_STREAM_JSON_VALUE  要加上，不然就会等待完成，一个请求体返回回去。

代码： 传送门