响应式编程库(三) -r2dbc

什么是r2dbc

反应式关系数据库连接（R2DBC）项目为关系数据库带来了反应式编程API。

基于Reactive Streams规范。R2DBC建立在Reactive Streams规范之上，它提供了一个完全反应式的非阻塞API。

r2dbc 官网：https://r2dbc.io/
github: r2dbc-mysql 版本
spring-data r2dbc

版本选择

参考下表来确定适合你的项目的r2 dbc-mysql版本。

spring-boot-starter-data-r2dbc	spring-data-r2dbc	r2dbc-spi	r2dbc-mysql(recommended)
3.0.* and above	3.0.* and above	1.0.0.RELEASE	io.asyncer:r2dbc-mysql:1.2.0
2.7.*	1.5.*	0.9.1.RELEASE	io.asyncer:r2dbc-mysql:0.9.7
2.6.* and below	1.4.* and below	0.8.6.RELEASE	dev.miku:r2dbc-mysql:0.8.2

简单试用

        <!-- https://mvnrepository.com/artifact/dev.miku/r2dbc-mysql --><dependency><groupId>dev.miku</groupId><artifactId>r2dbc-mysql</artifactId><version>0.8.2.RELEASE</version></dependency>

 @Testvoid connection() throws IOException {// r2dbc基于全异步、响应式、消息驱动// jdbc:mysql://localhost:3306/test// r2dbc:mysql://localhost:3306/test//0、MySQL配置MySqlConnectionConfiguration configuration = MySqlConnectionConfiguration.builder().host("192.168.xx.xx").port(3306).username("root").password("123456").database("test").connectTimeout(Duration.ofSeconds(3)).build();//1、获取连接工厂MySqlConnectionFactory connectionFactory = MySqlConnectionFactory.from(configuration);//2、获取到连接，发送sqlMono<Connection> connectionMono = Mono.from(connectionFactory.create());// JDBC： Statement： 封装sql的//3、数据发布者connectionMono.flatMapMany(connection ->connection
//                                    .createStatement("INSERT INTO `t_book` (`publisher`, `author`) VALUES ('who', '1')").createStatement("select * from t_book where id=?id and publisher=?").bind("id", 1L) //具名参数.bind(1, "pub").execute()).flatMap(result -> {// 不同版本，api有所不一致return result.map((readable,book)->{System.out.println("readable:"+readable);System.out.println("book:"+book);Long id = readable.get("id", Long.class);String publisher = readable.get("publisher", String.class);Long author = readable.get("author", Long.class);return new TBook(author,publisher,id);});}).subscribe(tAuthor -> System.out.println("book = " + tAuthor));//背压； 不用返回所有东西，基于请求量返回；System.in.read();}

结果：

readable:dev.miku.r2dbc.mysql.MySqlRow@34579a88
book:MySqlRowMetadata{metadata=[MySqlColumnMetadata{index=0, type=8, name='id', definitions=4203, nullability=NON_NULL, size=20, decimals=0, collationId=63}, MySqlColumnMetadata{index=1, type=253, name='publisher', definitions=1001, nullability=NON_NULL, size=1020, decimals=0, collationId=45}, MySqlColumnMetadata{index=2, type=3, name='author_id', definitions=1001, nullability=NON_NULL, size=11, decimals=0, collationId=63}, MySqlColumnMetadata{index=3, type=12, name='create_time', definitions=81, nullability=NON_NULL, size=19, decimals=0, collationId=63}], sortedNames=[author_id, create_time, id, publisher]}
book = TBook(authorId=1, publisher=pub, id=1)

整合springboot

        <!-- https://mvnrepository.com/artifact/io.asyncer/r2dbc-mysql --><dependency><groupId>io.asyncer</groupId><artifactId>r2dbc-mysql</artifactId><version>1.0.5</version></dependency><!--        响应式 Spring Data R2dbc--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-r2dbc</artifactId></dependency>

DatabaseClient 进行查询

    @Autowired  //贴近底层，join操作好做； 复杂查询好用DatabaseClient databaseClient; //数据库客户端@Testvoid databaseClient() throws IOException {// 底层操作databaseClient.sql("select * from t_author")
//                .bind(0,2L).fetch() //抓取数据.all()//返回所有.map(map -> {  //map == bean  属性=值System.out.println("map = " + map);String id = map.get("id").toString();String name = map.get("name").toString();return new TAuthor(Long.parseLong(id), name, null);}).subscribe(tAuthor -> System.out.println("tAuthor = " + tAuthor));System.in.read();}

spring:r2dbc:password: 123456username: rooturl: r2dbc:mysql://localhost:3306/testname: test

使用Repository接口(对应mapper)

/**
* TAuthor : 对应实体类；   Long  主键类型
*/
@Repository
public interface AuthorRepositories extends R2dbcRepository<TAuthor,Long> {//默认继承了一堆CRUD方法； 像mybatis-plus//QBC： Query By Criteria//QBE： Query By Example//成为一个起名工程师  where id In () and name like ?//仅限单表复杂条件查询。 不用编写sql！！！根据方法名自动生成sqlFlux<TAuthor> findAllByIdInAndNameLike(Collection<Long> id, String name);//多表复杂查询@Query("select * from t_author") //自定义query注解，指定sql语句Flux<TAuthor> findHaha();// 1-1关联关系； 查出这本图书以及它的作者@Query("select b.*,t.name as name from t_book b" +" LEFT JOIN t_author t on b.author_id = t.id " +" WHERE b.id = :bookId")Mono<TBookAuthor> authorBook(@Param("bookId") Long bookId);//    @Query("SELECT * FROM person WHERE lastname = :lastname")
//    Flux<Person> findByLastname(String lastname);
//
//    @Query("SELECT firstname, lastname FROM person WHERE lastname = $1")
//    Mono<Person> findFirstByLastname(String lastname);}

实体类

@Data
@AllArgsConstructor
@NoArgsConstructor@Table("t_book")
public class TBook {Long authorId;String publisher;@IdLong id;}

@Table("t_author")
@NoArgsConstructor
@AllArgsConstructor
@Data
public class TAuthor {@Idprivate Long id;private String name;//1-N如何封装@Transient //临时字段，并不是数据库表中的一个字段
//    @Field(exist=false)private List<TBook> TBooks;
}

@Table("t_book")
@Data
public class TBookAuthor {@IdLong id;Long authorId;String publisher;/*** 响应式中日期的映射用 Instant 或者 LocalXxx*/Instant createTime;/*** 一对一 关系 实体类*/TAuthor TAuthor;}

复杂查询（一对一）实体类转换器

springdata r2dbc mapping 文档

/*** 告诉Spring Data 怎么封装 TBookAuthor 对象*/
@ReadingConverter
public class BookAuthorConverter implements Converter<Row, TBookAuthor> {//1）、@Query 指定了 sql如何发送//2）、自定义 BookConverter 指定了 数据库返回的一 Row 数据，怎么封装成 TBook//3）、配置 R2dbcCustomConversions 组件，让 BookConverter 加入其中生效@Overridepublic TBookAuthor convert(Row source) {if(source == null) return null;//自定义结果集的封装TBookAuthor tBook = new TBookAuthor();tBook.setId(source.get("id", Long.class));tBook.setPublisher(source.get("publisher", String.class));Long author_id = source.get("author_id", Long.class);tBook.setAuthorId(author_id);tBook.setCreateTime(source.get("create_time", Instant.class));//让 converter兼容更多的表结构处理
//        if (source.get("name",String.class)) {TAuthor tAuthor = new TAuthor();tAuthor.setId(author_id);tAuthor.setName(source.get("name", String.class));tBook.setTAuthor(tAuthor);
//        }return tBook;}
}

注册转换器

@EnableR2dbcRepositories //开启 R2dbc 仓库功能；jpa
@Configuration
public class R2DbcConfiguration {@Bean //替换容器中原来的@ConditionalOnMissingBeanpublic R2dbcCustomConversions conversions(){//把我们的转换器加入进去； 效果新增了我们的 Converterreturn R2dbcCustomConversions.of(MySqlDialect.INSTANCE,new BookAuthorConverter());}
}

测试代码

@SpringBootTest
public class AppTest {@AutowiredTBookMapper bookMapper;@Testvoid testCRUD() throws IOException {
//         bookMapper.findAll().subscribe(System.out::println);
//        bookMapper.findById(1L).subscribe(System.out::println);TBookAuthor block = bookMapper.authorBook(1L).block();System.out.println(block);//查询是全异步的, 需要阻塞一下System.in.read();}}

一对多关系

    @Testvoid oneToN() throws IOException {//        databaseClient.sql("select a.id aid,a.name,b.* from t_author a  " +
//                "left join t_book b on a.id = b.author_id " +
//                "order by a.id")
//                .fetch()
//                .all(row -> {
//
//                })// 1~6// 1：false 2：false 3:false 4: true 8:true 5:false 6:false 7:false 8:true 9:false 10:false// [1,2,3]// [4,8]// [5,6,7]// [8]// [9,10]// bufferUntilChanged：// 如果下一个判定值比起上一个发生了变化就开一个新buffer保存，如果没有变化就保存到原buffer中//        Flux.just(1,2,3,4,8,5,6,7,8,9,10)
//                .bufferUntilChanged(integer -> integer%4==0 )
//                .subscribe(list-> System.out.println("list = " + list));; //自带分组Flux<TAuthor> flux = databaseClient.sql("select a.id aid,a.name,b.* from t_author a  " +"left join t_book b on a.id = b.author_id " +"order by a.id").fetch().all().bufferUntilChanged(rowMap -> Long.parseLong(rowMap.get("aid").toString())).map(list -> {TAuthor tAuthor = new TAuthor();Map<String, Object> map = list.get(0);tAuthor.setId(Long.parseLong(map.get("aid").toString()));tAuthor.setName(map.get("name").toString());//查到的所有图书List<TBook> tBooks = list.stream().map(ele -> {TBook tBook = new TBook();tBook.setId(Long.parseLong(ele.get("id").toString()));tBook.setAuthorId(Long.parseLong(ele.get("author_id").toString()));tBook.setTitle(ele.get("title").toString());return tBook;}).collect(Collectors.toList());tAuthor.setBooks(tBooks);return tAuthor;});//Long 数字缓存 -127 - 127；// 对象比较需要自己写好equals方法flux.subscribe(tAuthor -> System.out.println("tAuthor = " + tAuthor));System.in.read();}