Vert.x(vertx) 连接MySQL、Oracle数据库

快来打我* 2022-04-06 05:58 705阅读 0赞

Vert.x提供异步访问数据库的API，可能这里有朋友会有疑惑，直接使用我们之前的熟悉的Mybatis或者Hibernate不行吗，可行，但数据库操作是一个耗时操作，使用传统的同步模型，容易阻塞线程，导致整体性能下降，因此我们对于数据库操作，需要使用Vert.x提供的异步API。

Vert.x提供的API层级非常低，可以说是仅仅在原生JDBC基础上封装了一层异步接口。所有的对数据库操作都需要通过编写SQL来完成，参数的封装和结果的获取都需要手动的来实现，对于习惯使用ORM框架的开发者可能会非常的不习惯。

先来通过一个查询数据库的案例来演示如何使用Vert.x提供的异步API

## 基本操作 ##

1.引入数据库依赖，我们需要引入两个包，一个是vertx-jdbc，另一个是要真正连接数据库的驱动包，这里以MySQL为例

<dependency>
        <groupId>io.vertx</groupId>
        <artifactId>vertx-jdbc-client</artifactId>
        <version>3.6.0</version>
    </depend
    
    <dependency>
    	<groupId>mysql</groupId>
    	<artifactId>mysql-connector-java</artifactId>
    	<version>8.0.13</version>
    </dependency>

*注：2019-09-25 更新，在新的版本中 3.8.1，需要引入sql-common的包，否则会有类找不到。*

<dependency>
     <groupId>io.vertx</groupId>
     <artifactId>vertx-sql-common</artifactId>
     <version>3.8.1</version>
    </dependency>

2.抽象出一个DbUtils来方便获取数据库客户端，为了简单，直接就将配置写到代码里了

public class JdbcUtils {
    
            // 用于操作数据库的客户端
    	private JDBCClient dbClient;
    
    	public JdbcUtils(Vertx vertx) {
        
                    // 构造数据库的连接信息
    		JsonObject dbConfig = new JsonObject();
    		dbConfig.put("url", "jdbc:mysql://192.168.40.66:3306/test");
    		dbConfig.put("driver_class", "com.mysql.jdbc.Driver");
    		dbConfig.put("user", "xxxx");
    		dbConfig.put("password", "xxxx");
    
                    // 创建客户端
    		dbClient = JDBCClient.createShared(vertx, dbConfig);
    	}
    
            // 提供一个公共方法来获取客户端
    	public JDBCClient getDbClient() {
    		return dbClient;
    	}
    
    }

通过上面的工具类，可以快速的获取到客户端，看上面的代码也很简单，通过JsonObect构建一些基本的数据库连接信息，然后通过JDBCClient的createShard方法创建一个JDBCClient实例。

3.进行数据库的操作，以查询年龄大于18岁的用户为例

public class JdbcTestVerticle extends AbstractVerticle {
    
    	@Override
    	public void start() throws Exception {
    
    		// 获取到数据库连接的客户端
    		JDBCClient jdbcClient = new JdbcUtils(vertx).getDbClient();
    		String sql = "select * from t_user where age > ?";
    		// 构造参数
    		JsonArray params = new JsonArray().add(18);
    		// 执行查询
    		jdbcClient.queryWithParams(sql, params, qryRes->{
    			if(qryRes.succeeded()) {
    				// 获取到查询的结果，Vert.x对ResultSet进行了封装
    				ResultSet resultSet = qryRes.result();
    				// 把ResultSet转为List<JsonObject>形式
    				List<JsonObject> rows = resultSet.getRows();
    				// 输出结果
    				System.out.println(rows);
    			} else {
    				System.out.println("查询数据库出错！");
    			}
    		});
    	
    	}
    	
    	public static void main(String[] args) {
    		Vertx vertx = Vertx.vertx();
    		vertx.deployVerticle(new JdbcTestVerticle());
    	}
    	
    }

JsonArray是一个数组，SQL中用到的参数可以通过构建一个JsonArray来赋值。

JsonObejct是一个Json对象，类似于阿里的fastjson中提供的JSONObject

这两个对象在Vert.x中非常常用，而且非常的好用，但一定要注意空指针的问题，这是非常让人头疼的。

**优化**

通过上面的三个步骤，就可成功的对数据库进行操作了，但还有些问题需要优化，比如数据库连接信息放到配置文件中，再比如使用数据库连接池等等。

**\* 使用配置文件**

{
      "default":{
        "url":"jdbc:mysql://localhost:3306/my_project",
        "driver_class":"com.mysql.cj.jdbc.Driver",
        "user":"root",
        "password":"root"
      },
      "prod":{
        "url":"jdbc:mysql://localhost:3306/my_project",
        "driver_class":"com.mysql.cj.jdbc.Driver",
        "user":"root",
        "password":"root"
      }
    }

修改DbUtils工具类

public class JdbcUtils {
    
        private JDBCClient dbClient;
        private static JsonObject config ;
    
        static {
            byte[] buff = new byte[102400];
            try {
                // 读取配置文件
                InputStream ins = new FileInputStream("db.json");
                int i = IOUtils.read(ins, buff);
                config = new JsonObject(new String(buff, 0, i));
            } catch (Exception e) {
                System.out.println("读取配置文件失败");
            }
        }
    
        public JdbcUtils(Vertx vertx, String dsName) {
            JsonObject dbConfig = config.getJsonObject(dsName);
            if(dbConfig == null) {
                throw new RuntimeException("没有找到指定的数据源");
            }
            dbClient = JDBCClient.createShared(vertx, dbConfig);
        }
    
        public JdbcUtils(Vertx vertx) {
            this(vertx, "default");
        }
    
        public JDBCClient getDbClient() {
            return dbClient;
        }
    
    }

这样就支持了多个数据源，而且数据库连接配置都放到了配置文件中。

## **连接池配置** ##

数据连接池默认使用的C3P0，所以可以在db.json中进行配置C3P0连接池的参数就可以了，这里官网的地址为：[https://vertx.io/docs/vertx-jdbc-client/java/][https_vertx.io_docs_vertx-jdbc-client_java]

具体配置可以参考官网给出的配置，下面是一个简单的截图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tpbmdfa2do_size_16_color_FFFFFF_t_70][]

遗憾的是，Vert.x给出的数据库连接池的支持并不多，如果我们想要使用比如阿里的Druid连接池，需要自己来实现DataSourceProvider。当然DataSourceProvider的实现并不复杂，但麻烦啊！后面我会给出一个关于druid的DataSourceProvider的实现。

## **事务** ##

Vert.x从比较低的层面来控制事务，不像Spring一样可以使用声明式事务管理。要想在Vert.x中开启事务，和传统的JDBC管理事务的方式非常类似。首先要获得到连接，然后调用连接的setAutoCommit方法，关闭事务的自动提交，然后再手动的提交和回滚事务。

因为开启事务、提交事务、执行SQL都需要和数据库服务进行通信，因此在Vert.x中都是异步操作，按传统方式实现一个事务代码非常痛苦，看下面的一段开启事务的代码。写了一遍以后，绝对不愿意再写第二遍。

1. 获得连接

// 获得连接
    jdbcClient.getConnection(con -> {
        if (con.succeeded()) {
            System.out.println("获取到数据库连接");
    
            // 获取到的连接对象
            SQLConnection connection = con.result();
        }
    });

2. 设置不自动提交事务

// 开启事务
    connection.setAutoCommit(false, (v) -> {
        if (v.succeeded()) {
        
        }
    });

3.dml操作

// 执行更新操作
    connection.update("sql", upRes -> {
        if(upRes.succeed()){
        
        }
    });

4. 提交事务

// 提交事务
    connection.commit(rx -> {
        if (rx.succeeded()) {
            // 事务提交成功
        }
    });

回滚事务

// 回滚事务
    connection.rollback(rb -> {
        if (rb.succeeded()) {
            // 事务回滚成功
        }
    });

如果你觉得上面的还很简单，看看下面一个完整的例子吧，把这些嵌套在一起，你还觉得简单吗？

package stu.vertx.jdbc;
    
    import io.vertx.core.AbstractVerticle;
    import io.vertx.core.Vertx;
    import io.vertx.ext.jdbc.JDBCClient;
    import io.vertx.ext.sql.SQLConnection;
    
    /**
     * 获得数据库连接，执行查询，开启事务，执行更新操作
     *
     * @author <a href="https://blog.csdn.net/king_kgh>Kingh</a>
     * @version 1.0
     * @date 2019/4/3 9:19
     */
    public class GetConnection extends AbstractVerticle {
    
        @Override
        public void start() throws Exception {
    
            JDBCClient jdbcClient = new JdbcUtils(vertx).getDbClient();
            System.out.println("获取到数据库客户端");
            // 获取数据库连接
            jdbcClient.getConnection(con -> {
                if (con.succeeded()) {
                    System.out.println("获取到数据库连接");
    
                    // 获取到的连接对象
                    SQLConnection connection = con.result();
    
                    // 执行查询操作
                    connection.query("select * from t1", rs -> {
                        // 处理查询结果
                        if (rs.succeeded()) {
                            System.out.println(rs.result().getRows());
                        }
                    });
    
                    // 开启事务
                    connection.setAutoCommit(false, (v) -> {
                        if (v.succeeded()) {
                            // 事务开启成功 执行crud操作
                            connection.update("update t1 set name = '被修改了' where name = '111'", up -> {
    
                                if (up.succeeded()) {
                                    // 再来一笔写操作
                                    connection.update("insert into t1 values ('222','222222') ", up2 -> {
                                        if (up2.succeeded()) {
                                            // 提交事务
                                            connection.commit(rx -> {
                                                if (rx.succeeded()) {
                                                    // 事务提交成功
                                                }
                                            });
                                        } else {
                                            connection.rollback(rb -> {
                                                if (rb.succeeded()) {
                                                    // 事务回滚成功
                                                }
                                            });
                                        }
                                    });
                                } else {
                                    connection.rollback(rb -> {
                                        if (rb.succeeded()) {
                                            // 事务回滚成功
                                        }
                                    });
                                }
                            });
    
                        } else {
                            System.out.println("开启事务失败");
                        }
                    });
                } else {
                    System.out.println("获取数据库连接失败");
                }
            });
    
    
        }
    
        public static void main(String[] args) {
            Vertx.vertx().deployVerticle(new GetConnection());
        }
    }

## RxJava解决多层回调嵌套问题 ##

上面的代码仅仅是做了两个写操作，可以说是非常的痛苦了，一层一层的嵌套，根本没法维护。那么在真实的开发环境中，该如何管理事务呢，这就需要使用rxjava了，能够有效的减少多层嵌套带来的问题。使用rxjava首先是需要引入rxjava的依赖

<dependency>
     <groupId>io.vertx</groupId>
     <artifactId>vertx-rx-java</artifactId>
     <version>3.7.0</version>
    </dependency>

完成上面案例的同样代码如下

package stu.vertx.jdbc;
    
    import io.vertx.core.*;
    import io.vertx.core.json.JsonArray;
    import io.vertx.ext.jdbc.JDBCClient;
    import io.vertx.ext.sql.SQLConnection;
    import rx.Single;
    
    import java.util.UUID;
    
    /**
     * @author <a href="https://blog.csdn.net/king_kgh>Kingh</a>
     * @version 1.0
     * @date 2019/4/5 14:10
     */
    public class GetConnectionWithRxJava extends AbstractVerticle {
    
        @Override
        public void start() throws Exception {
    
            // 获取JDBC客户端
            JDBCClient jdbcClient = new JdbcUtils(vertx).getDbClient();
    
            getConnection(jdbcClient, con -> {
                if (con.succeeded()) {
                    // 获取到与数据库的连接
                    SQLConnection connection = con.result();
    
                    // 开启事务
                    rxOpenTx(connection)
                            // 执行写操作
                            .flatMap(this::rxExecuteUpdate1)
                            // 执行写操作
                            .flatMap(this::rxExecuteUpdate2)
                            .subscribe(ok -> {
                                // 提交事务
                                ok.commit(v -> {
                                });
                            }, err -> {
                                // 回滚事务
                                connection.rollback(v -> {
                                });
                            });
                }
            });
        }
    
        public Single<SQLConnection> rxOpenTx(SQLConnection connection) {
            return Single.create(new io.vertx.rx.java.SingleOnSubscribeAdapter<>(fut -> openTx(connection, fut)));
        }
    
        public Single<SQLConnection> rxExecuteUpdate1(SQLConnection connection) {
            return Single.create(new io.vertx.rx.java.SingleOnSubscribeAdapter<>(fut -> update1(connection, fut)));
        }
    
        public Single<SQLConnection> rxExecuteUpdate2(SQLConnection connection) {
            return Single.create(new io.vertx.rx.java.SingleOnSubscribeAdapter<>(fut -> update2(connection, fut)));
        }
    
        public void getConnection(JDBCClient jdbcClient, Handler<AsyncResult<SQLConnection>> resultHandler) {
            jdbcClient.getConnection(con -> {
                if (con.succeeded()) {
                    resultHandler.handle(Future.succeededFuture(con.result()));
                } else {
                    resultHandler.handle(Future.failedFuture(con.cause()));
                }
            });
        }
    
        public void openTx(SQLConnection connection, Handler<AsyncResult<SQLConnection>> resultHandler) {
            connection.setAutoCommit(false, o -> {
                if (o.succeeded()) {
                    resultHandler.handle(Future.succeededFuture(connection));
                } else {
                    resultHandler.handle(Future.failedFuture(o.cause()));
                }
            });
        }
    
        public void update1(SQLConnection connection, Handler<AsyncResult<SQLConnection>> resultHandler) {
            connection.updateWithParams("insert into t1 values (?,?)", new JsonArray().add(UUID.randomUUID().toString()).add(UUID.randomUUID().toString()), in -> {
                if (in.succeeded()) {
                    resultHandler.handle(Future.succeededFuture(connection));
                } else {
                    resultHandler.handle(Future.failedFuture(in.cause()));
                }
            });
        }
    
        public void update2(SQLConnection connection, Handler<AsyncResult<SQLConnection>> resultHandler) {
            connection.update("update t1 set name = '111' where passwd = '111'", in -> {
                if (in.succeeded()) {
                    resultHandler.handle(Future.succeededFuture(connection));
                } else {
                    resultHandler.handle(Future.failedFuture(in.cause()));
                }
            });
        }
    
        public static void main(String[] args) {
            Vertx.vertx().deployVerticle(new GetConnectionWithRxJava());
        }
    }

通过使用RxJava，没有那么深的嵌套层次，逻辑比较清晰。当然了，为了一个简单的操作，还是需要写很多的代码。

Vert.x相关系列文章

（一）Vert.x 简明介绍 [https://blog.csdn.net/king\_kgh/article/details/80772657][https_blog.csdn.net_king_kgh_article_details_80772657]

（二）Vert.x创建简单的HTTP服务 [https://blog.csdn.net/king\_kgh/article/details/80804078][https_blog.csdn.net_king_kgh_article_details_80804078]

（三）Vert.x Web开发之路由 [https://blog.csdn.net/king\_kgh/article/details/80848571][https_blog.csdn.net_king_kgh_article_details_80848571]

（四）Vert.x TCP服务实现 [https://blog.csdn.net/king\_kgh/article/details/84870775][https_blog.csdn.net_king_kgh_article_details_84870775]

（五）Vert.x数据库访问 [https://blog.csdn.net/king\_kgh/article/details/84894599][https_blog.csdn.net_king_kgh_article_details_84894599]

（六）Vert.x认证和授权 [https://blog.csdn.net/king\_kgh/article/details/85218454][https_blog.csdn.net_king_kgh_article_details_85218454]

（七）Vert.x事件总线（Event Bus）与远程服务调用 [https://blog.csdn.net/king\_kgh/article/details/86993812][https_blog.csdn.net_king_kgh_article_details_86993812]

Vert.x 案例代码：https://github.com/happy-fly

笔者就职于一家互联网支付公司，公司的核心项目使用的是Vert.x技术体系。记得笔者刚进入公司，接触Vert.x的时候，找遍了大大小小的网站，发现市面上关于Vert.x的文档除了官方文档外，几乎找不到其他资料。当时就励志要出一个专栏来写写Vert.x，以此帮助在Vert.x上采坑的朋友。因为笔者能力有限，文章中难免会有错误和疏漏，如果您对文章有意见或者好的建议，可以直接留言或者发送邮件到18366131816@163.com。如果您也对Vert.感兴趣，可以加入到我们，共同学习Vert.x，并推进国内开发者对Vert.x的认知。

[https_vertx.io_docs_vertx-jdbc-client_java]: https://vertx.io/docs/vertx-jdbc-client/java/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tpbmdfa2do_size_16_color_FFFFFF_t_70]: /images/20220406/1be98a0b0cc140528e93f5d905716e19.png
[https_blog.csdn.net_king_kgh_article_details_80772657]: https://blog.csdn.net/king_kgh/article/details/80772657
[https_blog.csdn.net_king_kgh_article_details_80804078]: https://blog.csdn.net/king_kgh/article/details/80804078
[https_blog.csdn.net_king_kgh_article_details_80848571]: https://blog.csdn.net/king_kgh/article/details/80848571
[https_blog.csdn.net_king_kgh_article_details_84870775]: https://blog.csdn.net/king_kgh/article/details/84870775
[https_blog.csdn.net_king_kgh_article_details_84894599]: https://blog.csdn.net/king_kgh/article/details/84894599
[https_blog.csdn.net_king_kgh_article_details_85218454]: https://blog.csdn.net/king_kgh/article/details/85218454
[https_blog.csdn.net_king_kgh_article_details_86993812]: https://blog.csdn.net/king_kgh/article/details/86993812