响应式关系数据库处理R2DBC
文章目录
- 简介
- R2DBC介绍
- 项目依赖
- 创建ConnectionFactory
- 创建Entity Bean
- 初始化数据库
- 获取所有的用户
- Prepare Statement
- 事务处理
- WebFlux使用
- 执行效果
简介
之前我们提到过,对于底层的数据源来说,MongoDB, Redis, 和 Cassandra 可以直接以reactive的方式支持Spring Data。而其他很多关系型数据库比如Postgres, Microsoft SQL Server, MySQL, H2 和 Google Spanner 则可以通过使用R2DBC 来实现对reactive的支持。
今天我们就来具体讲解一下R2DBC的使用。
R2DBC介绍
之前我们介绍了Reactor还有基于其之上的Spring WebFlux框架。包括vert.x,rxjava等等reactive技术。我们实际上在应用层已经有很多优秀的响应式处理框架。
但是有一个问题就是所有的框架都需要获取底层的数据,而基本上关系型数据库的底层读写都还是同步的。
为了解决这个问题,出现了两个标准,一个是oracle提出的 ADBC (Asynchronous Database Access API),另一个就是Pivotal提出的R2DBC (Reactive Relational Database Connectivity)。
R2DBC是基于Reactive Streams标准来设计的。通过使用R2DBC,你可以使用reactive API来操作数据。
同时R2DBC只是一个开放的标准,而各个具体的数据库连接实现,需要实现这个标准。
今天我们以r2dbc-h2为例,讲解一下r2dbc在Spring webFlux中的使用。
项目依赖
我们需要引入r2dbc-spi和r2dbc-h2两个库,其中r2dbc-spi是接口,而r2dbc-h2是具体的实现。
同时我们使用了Spring webflux,所以还需要引入spring-boot-starter-webflux。
具体的依赖如下:
<!-- R2DBC H2 Driver --><dependency><groupId>io.r2dbc</groupId><artifactId>r2dbc-h2</artifactId><version>${r2dbc-h2.version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency>
创建ConnectionFactory
ConnectionFactory是数据库连接的一个具体实现,通过ConnectionFactory我们可以创建到数据库的连接。
先看一下数据库的配置文件,为了方便起见,这里我们使用的是内存数据库H2 :
r2dbc.url=r2dbc:h2:mem://./r2dbc
r2dbc.user=sa
r2dbc.password=password
第一个url指定的是数据库的连接方式,下面两个是数据库的用户名和密码。
接下来我们看一下,怎么通过这些属性来创建ConnectionFactory:
@Beanpublic ConnectionFactory connectionFactory() {ConnectionFactoryOptions baseOptions = ConnectionFactoryOptions.parse(url);ConnectionFactoryOptions.Builder ob = ConnectionFactoryOptions.builder().from(baseOptions);if (!StringUtil.isNullOrEmpty(user)) {ob = ob.option(USER, user);}if (!StringUtil.isNullOrEmpty(password)) {ob = ob.option(PASSWORD, password);}return ConnectionFactories.get(ob.build());}
通过url可以parse得到ConnectionFactoryOptions。然后通过ConnectionFactories的get方法创建ConnectionFactory。
如果我们设置了USER或者PASSWORD,还可以加上这两个配置。
创建Entity Bean
这里,我们创建一个简单的User对象:
@Data
@NoArgsConstructor
@AllArgsConstructor
public class Users {private Long id;private String firstname;private String lastname;
}
初始化数据库
虽然H5有很多更加简单的方式来初始化数据库,比如直接读取SQL文件,这里为了说明R2DBC的使用,我们使用手动的方式来创建:
@Beanpublic CommandLineRunner initDatabase(ConnectionFactory cf) {return (args) ->Flux.from(cf.create()).flatMap(c ->Flux.from(c.createBatch().add("drop table if exists Users").add("create table Users(" +"id IDENTITY(1,1)," +"firstname varchar(80) not null," +"lastname varchar(80) not null)").add("insert into Users(firstname,lastname)" +"values('flydean','ma')").add("insert into Users(firstname,lastname)" +"values('jacken','yu')").execute()).doFinally((st) -> c.close())).log().blockLast();}
上面的代码中,我们使用c.createBatch()来向数据库插入一些数据。
除了createBatch,还可以使用create来创建单个的执行语句。
获取所有的用户
在Dao中,我们提供了一个findAll的方法:
public Flux<Users> findAll() {return Mono.from(connectionFactory.create()).flatMap((c) -> Mono.from(c.createStatement("select id,firstname,lastname from users").execute()).doFinally((st) -> close(c))).flatMapMany(result -> Flux.from(result.map((row, meta) -> {Users acc = new Users();acc.setId(row.get("id", Long.class));acc.setFirstname(row.get("firstname", String.class));acc.setLastname(row.get("lastname", String.class));return acc;})));}
简单解释一下上面的使用。
因为是一个findAll方法,我们需要找出所有的用户信息。所以我们返回的是一个Flux而不是一个Mono。
怎么从Mono转换成为一个Flux呢?
这里我们使用的是flatMapMany,将select出来的结果,分成一行一行的,最后转换成为Flux。
Prepare Statement
为了防止SQL注入,我们需要在SQL中使用Prepare statement:
public Mono<Users> findById(long id) {return Mono.from(connectionFactory.create()).flatMap(c -> Mono.from(c.createStatement("select id,firstname,lastname from Users where id = $1").bind("$1", id).execute()).doFinally((st) -> close(c))).map(result -> result.map((row, meta) ->new Users(row.get("id", Long.class),row.get("firstname", String.class),row.get("lastname", String.class)))).flatMap( p -> Mono.from(p));}
看下我们是怎么在R2DBC中使用prepare statement的。
事务处理
接下来我们看一下怎么在R2DBC中使用事务:
public Mono<Users> createAccount(Users account) {return Mono.from(connectionFactory.create()).flatMap(c -> Mono.from(c.beginTransaction()).then(Mono.from(c.createStatement("insert into Users(firstname,lastname) values($1,$2)").bind("$1", account.getFirstname()).bind("$2", account.getLastname()).returnGeneratedValues("id").execute())).map(result -> result.map((row, meta) ->new Users(row.get("id", Long.class),account.getFirstname(),account.getLastname()))).flatMap(pub -> Mono.from(pub)).delayUntil(r -> c.commitTransaction()).doFinally((st) -> c.close()));}
上面的代码中,我们使用了事务,具体的代码有两部分:
c -> Mono.from(c.beginTransaction())
.delayUntil(r -> c.commitTransaction())
开启是的时候需要使用beginTransaction,后面提交就需要调用commitTransaction。
WebFlux使用
最后,我们需要创建WebFlux应用来对外提供服务:
@GetMapping("/users/{id}")public Mono<ResponseEntity<Users>> getUsers(@PathVariable("id") Long id) {return usersDao.findById(id).map(acc -> new ResponseEntity<>(acc, HttpStatus.OK)).switchIfEmpty(Mono.just(new ResponseEntity<>(null, HttpStatus.NOT_FOUND)));}@GetMapping("/users")public Flux<Users> getAllAccounts() {return usersDao.findAll();}@PostMapping("/createUser")public Mono<ResponseEntity<Users>> createUser(@RequestBody Users user) {return usersDao.createAccount(user).map(acc -> new ResponseEntity<>(acc, HttpStatus.CREATED)).log();}
执行效果
最后,我们运行一下代码,执行下users:
curl "localhost:8080/users"
[{"id":1,"firstname":"flydean","lastname":"ma"},{"id":2,"firstname":"jacken","lastname":"yu"}]%
完美,实验成功。
本文的代码:webflux-with-r2dbc
本文作者:flydean程序那些事
本文链接:http://www.flydean.com/r2dbc-introduce/
本文来源:flydean的博客
欢迎关注我的公众号:「程序那些事」最通俗的解读,最深刻的干货,最简洁的教程,众多你不知道的小技巧等你来发现!
响应式关系数据库处理R2DBC相关推荐
- r2dbc mysql_使用R2DBC实现数据库的响应式访问
Reactive Programming可以看作是一种编程模型,它通过创建事件驱动的非阻塞功能管道来促进可扩展性和稳定性,这些管道对资源的可用性和可处理性做出反应.延迟执行, 并发和异步性 只是底层编 ...
- mysql as用法_Spring Data R2DBC响应式操作MySQL
1. 前言 在使用 R2DBC 操作 MySQL 数据库 一文中初步介绍了r2dbc-mysql的使用.但是借助于DatabaseClient操作MySQL,过于初级和底层,不利于开发.今天就利用Sp ...
- response获取响应内容_Project Reactor 深度解析 - 1. 响应式编程介绍,实现以及现有问题
现在, Java 的各种基于 Reactor 模型的响应式编程库或者框架越来越多了,像是 RxJava,Project Reactor,Vert.x 等等等等.在 Java 9, Java 也引入了自 ...
- 响应式圣经:10W字,实现Spring响应式编程自由
前言 全链路异步化改造的基础是响应式编程 随着业务的发展,微服务应用的流量越来越大,使用到的资源也越来越多. 在微服务架构下,大量的应用都是 SpringCloud 分布式架构,这种架构总体上是全链路 ...
- Reactive 响应式编程简单使用
Reactive Stream 模型 了解reactive stream(Flow类) 在介绍java版本的reactive stream之前,我们先回顾一下reactive stream需要做哪些事 ...
- 我对响应式编程中Mono和Flux的理解
1. 前言 很多同学反映对响应式编程中的Flux和Mono这两个Reactor中的概念有点懵逼.但是目前Java响应式编程中我们对这两个对象的接触又最多,诸如Spring WebFlux.RSocke ...
- 响应式编程之Spring Webflux
文章目录 一 .响应式编程 二 .响应式流 (1)JDK9响应式流: (2)Reactor响应式流库 三.Spring WebFlux 1.整合Webflux 2.事件推送 3.实现背压 四.配置数据 ...
- 艿艿连肝了几个周末,写了一篇贼长的 Spring 响应式 Web 框架 WebFlux!市面第二完整~...
点击上方"芋道源码",选择"设为星标" 做积极的人,而不是积极废人! 源码精品专栏 原创 | Java 2020 超神之路,很肝~ 中文详细注释的开源项目 RP ...
- RSocket一种新的响应式应用新协议
简介 RSocket是在华盛顿特区举行的SpringOne平台会议上宣布的,是一种新的第7层语言无关的应用网络协议.它是一种基于Reactive Streams背压的双向,多路复用,基于消息的二进制协 ...
最新文章
- spring+mybatis事务的readonly属性无效
- Linux 批量依赖库拷贝(ldd)
- Ubuntu终端命令行不显示颜色
- 米哈游CEO蔡浩宇GDC演讲:《原神》如何打造动人的开放世界?
- 安卓如何调出软键盘_智能汽车到底如何交互?小鹏用全场景语音给出了答案
- windows下实现自己的第一个python脚本文件并.exe运行
- Log4j框架配置文件
- Spring boot - 整合 Redis缓存(上)
- android开发之android:padding和android:margin的区别
- Python、PyCharm、Django框架安装
- Linux du 命令
- php写2048,原生js编写2048小游戏实例代码
- 期刊分类abcde_收藏!中文顶级期刊分类,多图教你不被ABCD类搞晕!
- CentOS怎样添加字体
- XVII Open Cup Eastern Grand Prix - J Votter and Paul De Mort
- 对当下很火的两大短视频平台 抖音 和 微视进行竞品分析
- C语言核心知识点总结
- 辰视将携3D视觉技术及各领域解决方案参加华南工博会国际机器视觉展
- tl494c封装区别_TL494芯片详细资料
- Linux时间编程三大步骤