深入理解r2dbc在mysql中的使用

• 生活家

简介

mysql应该是我们在日常工作中使用到的一个非常普遍的数据库，虽然mysql现在是oracle公司的，但是它是开源的，市场占有率还是非常高的。

今天我们将会介绍r2dbc在mysql中的使用。

r2dbc-mysql的maven依赖

要想使用r2dbc-mysql，我们需要添加如下的maven依赖：

<dependency>
  <groupid>dev.miku</groupid>
  <artifactid>r2dbc-mysql</artifactid>
  <version>0.8.2.release</version>
</dependency>

当然，如果你想使用snapshot版本的话，可以这样：

<dependency>
  <groupid>dev.miku</groupid>
  <artifactid>r2dbc-mysql</artifactid>
  <version>${r2dbc-mysql.version}.build-snapshot</version>
</dependency>

<repository>
  <id>sonatype-snapshots</id>
  <name>sonatype snapshots</name>
  <url>https://oss.sonatype.org/content/repositories/snapshots</url>
  <snapshots>
    <enabled>true</enabled>
  </snapshots>
</repository>

创建connectionfactory

创建connectionfactory的代码实际上使用的r2dbc的标准接口，所以和之前讲到的h2的创建代码基本上是一样的：

// notice: the query string must be url encoded
connectionfactory connectionfactory = connectionfactories.get(
  "r2dbcs:mysql://root:database-password-in-here@127.0.0.1:3306/r2dbc?" +
  "zerodate=use_round&" +
  "sslmode=verify_identity&" +
  "useserverpreparestatement=true&" +
  "tlsversion=tlsv1.3%2ctlsv1.2%2ctlsv1.1&" +
  "sslca=%2fpath%2fto%2fmysql%2fca.pem&" +
  "sslkey=%2fpath%2fto%2fmysql%2fclient-key.pem&" +
  "sslcert=%2fpath%2fto%2fmysql%2fclient-cert.pem&" +
  "sslkeypassword=key-pem-password-in-here"
)

// creating a mono using project reactor
mono<connection> connectionmono = mono.from(connectionfactory.create());

不同的是connectionfactories传入的参数不同。

我们也支持unix domain socket的格式：

// minimum configuration for unix domain socket
connectionfactory connectionfactory = connectionfactories.get("r2dbc:mysql://root@unix?unixsocket=%2fpath%2fto%2fmysql.sock")

mono<connection> connectionmono = mono.from(connectionfactory.create());

同样的，我们也支持从connectionfactoryoptions中创建connectionfactory：

connectionfactoryoptions options = connectionfactoryoptions.builder()
  .option(driver, "mysql")
  .option(host, "127.0.0.1")
  .option(user, "root")
  .option(port, 3306) // optional, default 3306
  .option(password, "database-password-in-here") // optional, default null, null means has no password
  .option(database, "r2dbc") // optional, default null, null means not specifying the database
  .option(connect_timeout, duration.ofseconds(3)) // optional, default null, null means no timeout
  .option(ssl, true) // optional, default sslmode is "preferred", it will be ignore if sslmode is set
  .option(option.valueof("sslmode"), "verify_identity") // optional, default "preferred"
  .option(option.valueof("sslca"), "/path/to/mysql/ca.pem") // required when sslmode is verify_ca or verify_identity, default null, null means has no server ca cert
  .option(option.valueof("sslcert"), "/path/to/mysql/client-cert.pem") // optional, default null, null means has no client cert
  .option(option.valueof("sslkey"), "/path/to/mysql/client-key.pem") // optional, default null, null means has no client key
  .option(option.valueof("sslkeypassword"), "key-pem-password-in-here") // optional, default null, null means has no password for client key (i.e. "sslkey")
  .option(option.valueof("tlsversion"), "tlsv1.3,tlsv1.2,tlsv1.1") // optional, default is auto-selected by the server
  .option(option.valueof("sslhostnameverifier"), "com.example.demo.myverifier") // optional, default is null, null means use standard verifier
  .option(option.valueof("sslcontextbuildercustomizer"), "com.example.demo.mycustomizer") // optional, default is no-op customizer
  .option(option.valueof("zerodate"), "use_null") // optional, default "use_null"
  .option(option.valueof("useserverpreparestatement"), true) // optional, default false
  .option(option.valueof("tcpkeepalive"), true) // optional, default false
  .option(option.valueof("tcpnodelay"), true) // optional, default false
  .option(option.valueof("autodetectextensions"), false) // optional, default false
  .build();
connectionfactory connectionfactory = connectionfactories.get(options);

// creating a mono using project reactor
mono<connection> connectionmono = mono.from(connectionfactory.create());

或者下面的unix domain socket格式：

// minimum configuration for unix domain socket
connectionfactoryoptions options = connectionfactoryoptions.builder()
  .option(driver, "mysql")
  .option(option.valueof("unixsocket"), "/path/to/mysql.sock")
  .option(user, "root")
  .build();
connectionfactory connectionfactory = connectionfactories.get(options);

mono<connection> connectionmono = mono.from(connectionfactory.create());

使用mysqlconnectionfactory创建connection

上面的例子中，我们使用的是通用的r2dbc api来创建connection，同样的，我们也可以使用特有的mysqlconnectionfactory来创建connection：

mysqlconnectionconfiguration configuration = mysqlconnectionconfiguration.builder()
  .host("127.0.0.1")
  .user("root")
  .port(3306) // optional, default 3306
  .password("database-password-in-here") // optional, default null, null means has no password
  .database("r2dbc") // optional, default null, null means not specifying the database
  .serverzoneid(zoneid.of("continent/city")) // optional, default null, null means query server time zone when connection init
  .connecttimeout(duration.ofseconds(3)) // optional, default null, null means no timeout
  .sslmode(sslmode.verify_identity) // optional, default sslmode.preferred
  .sslca("/path/to/mysql/ca.pem") // required when sslmode is verify_ca or verify_identity, default null, null means has no server ca cert
  .sslcert("/path/to/mysql/client-cert.pem") // optional, default has no client ssl certificate
  .sslkey("/path/to/mysql/client-key.pem") // optional, default has no client ssl key
  .sslkeypassword("key-pem-password-in-here") // optional, default has no client ssl key password
  .tlsversion(tlsversions.tls1_3, tlsversions.tls1_2, tlsversions.tls1_1) // optional, default is auto-selected by the server
  .sslhostnameverifier(myverifier.instance) // optional, default is null, null means use standard verifier
  .sslcontextbuildercustomizer(mycustomizer.instance) // optional, default is no-op customizer
  .zerodateoption(zerodateoption.use_null) // optional, default zerodateoption.use_null
  .useserverpreparestatement() // use server-preparing statements, default use client-preparing statements
  .tcpkeepalive(true) // optional, controls tcp keep alive, default is false
  .tcpnodelay(true) // optional, controls tcp no delay, default is false
  .autodetectextensions(false) // optional, controls extension auto-detect, default is true
  .extendwith(myextension.instance) // optional, manual extend an extension into extensions, default using auto-detect
  .build();
connectionfactory connectionfactory = mysqlconnectionfactory.from(configuration);

// creating a mono using project reactor
mono<connection> connectionmono = mono.from(connectionfactory.create());

或者下面的unix domain socket方式：

// minimum configuration for unix domain socket
mysqlconnectionconfiguration configuration = mysqlconnectionconfiguration.builder()
  .unixsocket("/path/to/mysql.sock")
  .user("root")
  .build();
connectionfactory connectionfactory = mysqlconnectionfactory.from(configuration);

mono<connection> connectionmono = mono.from(connectionfactory.create());

执行statement

首先看一个简单的不带参数的statement:

connection.createstatement("insert into `person` (`first_name`, `last_name`) values ('who', 'how')")
  .execute(); // return a publisher include one result

然后看一个带参数的statement:

connection.createstatement("insert into `person` (`birth`, `nickname`, `show_name`) values (?, ?name, ?name)")
  .bind(0, localdatetime.of(2019, 6, 25, 12, 12, 12))
  .bind("name", "some one") // not one-to-one binding, call twice of native index-bindings, or call once of name-bindings.
  .add()
  .bind(0, localdatetime.of(2009, 6, 25, 12, 12, 12))
  .bind(1, "my nickname")
  .bind(2, "naming show")
  .returngeneratedvalues("generated_id")
  .execute(); // return a publisher include two results.

注意，如果参数是null的话，可以使用bindnull来进行null值的绑定。

接下来我们看一个批量执行的操作：

connection.createbatch()
  .add("insert into `person` (`first_name`, `last_name`) values ('who', 'how')")
  .add("update `earth` set `count` = `count` + 1 where `id` = 'human'")
  .execute(); // return a publisher include two results.

执行事务

我们看一个执行事务的例子：

connection.begintransaction()
  .then(mono.from(connection.createstatement("insert into `person` (`first_name`, `last_name`) values ('who', 'how')").execute()))
  .flatmap(result::getrowsupdated)
  .thenmany(connection.createstatement("insert into `person` (`birth`, `nickname`, `show_name`) values (?, ?name, ?name)")
    .bind(0, localdatetime.of(2019, 6, 25, 12, 12, 12))
    .bind("name", "some one")
    .add()
    .bind(0, localdatetime.of(2009, 6, 25, 12, 12, 12))
    .bind(1, "my nickname")
    .bind(2, "naming show")
    .returngeneratedvalues("generated_id")
    .execute())
  .flatmap(result::getrowsupdated)
  .then(connection.committransaction());

使用线程池

为了提升数据库的执行效率，减少建立连接的开销，一般数据库连接都会有连接池的概念，同样的r2dbc也有一个叫做r2dbc-pool的连接池。

r2dbc-pool的依赖：

<dependency>
 <groupid>io.r2dbc</groupid>
 <artifactid>r2dbc-pool</artifactid>
 <version>${version}</version>
</dependency>

如果你想使用snapshot版本，也可以这样指定：

<dependency>
 <groupid>io.r2dbc</groupid>
 <artifactid>r2dbc-pool</artifactid>
 <version>${version}.build-snapshot</version>
</dependency>

<repository>
 <id>spring-libs-snapshot</id>
 <name>spring snapshot repository</name>
 <url>https://repo.spring.io/libs-snapshot</url>
</repository>

看一下怎么指定数据库连接池：

connectionfactory connectionfactory = connectionfactories.get("r2dbc:pool:<my-driver>://<host>:<port>/<database>[?maxidletime=pt60s[&…]");

publisher<? extends connection> connectionpublisher = connectionfactory.create();

可以看到，我们只需要在连接url上面添加pool这个driver即可。

同样的，我们也可以通过connectionfactoryoptions来创建：

connectionfactory connectionfactory = connectionfactories.get(connectionfactoryoptions.builder()
  .option(driver, "pool")
  .option(protocol, "postgresql") // driver identifier, protocol is delegated as driver by the pool.
  .option(host, "…")
  .option(port, "…") 
  .option(user, "…")
  .option(password, "…")
  .option(database, "…")
  .build());

publisher<? extends connection> connectionpublisher = connectionfactory.create();

// alternative: creating a mono using project reactor
mono<connection> connectionmono = mono.from(connectionfactory.create());

最后， 你也可以直接通过创建connectionpoolconfiguration来使用线程池：

connectionfactory connectionfactory = …;

connectionpoolconfiguration configuration = connectionpoolconfiguration.builder(connectionfactory)
  .maxidletime(duration.ofmillis(1000))
  .maxsize(20)
  .build();

connectionpool pool = new connectionpool(configuration);
 

mono<connection> connectionmono = pool.create();

// later

connection connection = …;
mono<void> release = connection.close(); // released the connection back to the pool

// application shutdown
pool.dispose();

到此这篇关于深入理解r2dbc在mysql中的使用的文章就介绍到这了,更多相关mysql r2dbc 内容请搜索www.887551.com以前的文章或继续浏览下面的相关文章希望大家以后多多支持www.887551.com！

赞 (0)
上一篇 2022年3月21日
下一篇 2022年3月21日

相关推荐