Vert.x MySQL数据库基本操作

教程分享 > Java教程 > Vert.x (519) 2024-08-14 14:20:00

前言

本文主要讲解在Vert.x环境下与Mysql数据库连接和操作，实现基础的增删改查和事务，以及REST接口返回数据库中查询结果。项目引用之前的《Vert.x 4 Web应用初识》

环境依赖

MySQL数据库，本文使用版本5.7，数据库安装和配置不在这里讲述。

创建一个数据库 vertx 字符集选utfmb4

并创建一个用户信息表，表创建参考：

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for user_info
-- ----------------------------
DROP TABLE IF EXISTS `user_info`;
CREATE TABLE `user_info`  (
 `id` int(11) NOT NULL AUTO_INCREMENT,
 `username` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
 `address` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
 `age` int(11) NULL DEFAULT NULL,
 `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
 PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 10 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
-- ----------------------------
-- Records of user_info
-- ----------------------------
INSERT INTO `user_info` VALUES (1, '张三', '成都', 23, '2024-08-14 10:35:39');
INSERT INTO `user_info` VALUES (2, '李四', '北京', 21, '2024-08-14 11:35:39');
INSERT INTO `user_info` VALUES (3, '王五', '云南', 23, '2024-08-14 11:35:57');
SET FOREIGN_KEY_CHECKS = 1;

引入数据库相关依赖

maven pom.xml

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-mysql-client</artifactId>
  <version>4.5.9</version>
</dependency>

Vert.x MySQL client数据库使用

创建数据库操作需要的连接池

代码聚焦

    MySQLConnectOptions connectOptions = new MySQLConnectOptions()
      .setDatabase("vertx")
      .setHost("127.0.0.1")
      .setPort(3306)
      .setUser("root")
      .setPassword("root");
    PoolOptions poolOptions = new PoolOptions()
      .setMaxSize(5);
    Pool pool =  Pool.pool(vertx, connectOptions, poolOptions);
//    Pool pool = MySQLBuilder.pool()
//      .with(poolOptions)
//      .connectingTo(connectOptions)
//      .build();

注意，在vertx context环境需要使用vertx方式构建pool,否则报错。

数据库查询示例

数据库查询包含普通查询和传参查询

代码聚焦

private void query(Pool pool){
  //查询
  pool.getConnection().compose(connection -> connection
    .query("select * from user_info where id = 1")
    .execute()
    .onComplete(ar->{
      if(ar.succeeded()){
        RowSet<Row> result = ar.result();
        log.info("result: {}", result.size());
        for (Row row : result) {
          JsonObject json = row.toJson();
          log.info("json one : {}", json);
        }
      }else{
        log.error("error", ar.cause());
      }
    }));

  //参数查询
  Integer id = 2;
  pool
    .preparedQuery("select * from user_info where id = ? and age > ?")
    .execute(Tuple.of(id,10))
    .onComplete(ar->{
      if(ar.succeeded()){
        RowSet<Row> result = ar.result();
        log.info("result: {}", result.size());
        for (Row row : result) {
          JsonObject json = row.toJson();
          log.info("preparedQuery json one : {}", json);
        }
      }else {
        log.error("error", ar.cause());
      }
    });
}

启动项目在控制台可以看到日志信息：

Vert.x MySQL数据库基本操作_图示-c15c23d6d8fc4761826aa7cd1b535cb4.png — 控制台查询数据日志

从上图可以看到输出结果与我们预期结果一致。

新增数据并获取自增长id值

代码聚焦

private void insert(Pool pool){
  pool
    .preparedQuery("insert into user_info (username,address,age) values (?,?,?)")
    .execute(Tuple.of("刘三","广西",29))
    .onComplete(ar->{
      if(ar.succeeded()){
        RowSet<Row> result = ar.result();
        log.info("result rowCount: {}", result.rowCount());

        //新增数据id
        long lastInsertId = result.property(MySQLClient.LAST_INSERTED_ID);
        log.info("lastInsertId: {}", lastInsertId);

      }else {
        log.error("error", ar.cause());
      }
    });
}

启动项目观察控制台信息

Vert.x MySQL数据库基本操作_图示-15d3ada3699b4fbda37ba6799d96dd34.png — 插入数据控制台日志

Vert.x MySQL数据库基本操作_图示-8884652b8c394f7fb45a337b018f6d1e.png — 数据库查看

可以看到插入的数据id为4与预期结果一致

批量操作

批量插入示例

代码聚焦

private void insertBatch(Pool pool){
  //批量新增
  List<Tuple> batchDataList = new ArrayList<>();
  batchDataList.add(Tuple.of("刘四","上海",23));
  batchDataList.add(Tuple.of("唐五","苏州",33));
  pool
    .preparedQuery("insert into user_info (username,address,age) values (?,?,?)")
    .executeBatch(batchDataList)
    .onComplete(res->{
      if(res.succeeded()){
        RowSet<Row> result = res.result();
        log.info("result rowCount: {}", result.rowCount());
      }else {
        log.error("error", res.cause());
      }
    });
}

操作完成后查看数据库

Vert.x MySQL数据库基本操作_图示-a0d35361375246e3bcb8c23db1ed6fbd.png — 批量操作结果

从数据库可以看到结果和我们预期一致。

删除数据操作

代码聚焦

private void delete(Pool pool){
  pool.preparedQuery("delete from user_info where id = ?")
    .execute(Tuple.of(1))
    .onComplete(ar->{
      if(ar.succeeded()){
        RowSet<Row> result = ar.result();
        log.info("result rowCount: {}", result.rowCount());
      }else{
        log.error("error", ar.cause());
      }
    });
}

启动项目然后去查看数据库信息

Vert.x MySQL数据库基本操作_图示-0d5ae3069adf4de89f495967b98b831f.png

从数据库信息来看，结果与我们预期一致。

事务操作

事务操作在业务处理很常见，下面是vert.x mysql client的事务操作示例

代码聚焦

/**
 * 事务测试
 * @param pool
 */
private void transactions(Pool pool){
  pool.getConnection()
    .onSuccess(conn->{ //连接
      conn.begin()//开启事务
        .compose(tx->conn
          .query("INSERT INTO user_info (username,age) VALUES ('皮5',32)")
          .execute()
          .compose(res2->conn
            .query("INSERT INTO user_info (username,age) VALUES (null,23)")
            .execute())
          .compose(res3->tx.commit())//提交事务
        .eventually(v->conn.close())
        .onSuccess(v->{
          log.info("transactions success");
        })
        .onFailure(err->{
          log.error("transactions failure: {}", err.getMessage());
        }));
    });
}

上方代码在一个连接执行了两个操作，第一个操作是没问题的，第二个操作由于给username传入了null违背了数据库定义的非空策略导致报错，最终事务回滚两个都步成功。可启动项目观察控制台输出

Vert.x MySQL数据库基本操作_图示-e3057664a2f345c4a988658c217edc95.png

控制台信息输出错误，并且去数据库查看并未插入任何数据，与事务逻辑预期结果一致。

另外事务操作也可以如下简化编写

/**
 * 事务简化使用
 * @param pool
 */
private void transactionsSimplified(Pool pool){

  pool.withTransaction(conn->conn.query("INSERT INTO user_info (username,age) VALUES ('皮5',32)")
    .execute()
    .flatMap(res->conn
      .query("INSERT INTO user_info (username,age) VALUES (null,23)")
      .execute()
      .map("Users inserted"))
    .onSuccess(v->{
      log.info("transactions success");
    })
    .onFailure(err->{
      log.error("transactions failure: {}", err.getMessage());
    }));
}

数据库查询结果配合HTTP REST接口返回

查询接口定义

  private void queryById(Pool pool,Integer id, Handler<AsyncResult<JsonObject>> resultHandler){
    //参数查询
    pool
      .preparedQuery("select * from user_info where id = ? and age > ?")
      .execute(Tuple.of(id,10))
      .onComplete(ar->{
        if(ar.succeeded()){
          RowSet<Row> result = ar.result();
          log.info("result: {}", result.size());
          if (result.size() > 0) {
            Row row = result.iterator().next();
            JsonObject json = row.toJson();
            log.info("preparedQuery json one : {}", json);
            resultHandler.handle(Future.succeededFuture(json));
          }else {
            resultHandler.handle(Future.failedFuture("no row"));
//            resultHandler.handle(Future.succeededFuture(null));
          }

        }else {
          log.error("error", ar.cause());
          resultHandler.handle(Future.failedFuture(ar.cause().getMessage()));
        }
      });
  }

HTTP REST 接口调用查询结果

    router.get("/user/:id")
      .handler(ctx->{
        String id = ctx.request().getParam("id");
        queryById(pool, Integer.parseInt(id), jsonObjectAsyncResult -> {
          if (jsonObjectAsyncResult.succeeded()) {
            JsonObject result = jsonObjectAsyncResult.result();
            ctx.json(result);
          }else {
//            ctx.fail(jsonObjectAsyncResult.cause());
            ctx.response()
              .setStatusCode(500)
              .end(jsonObjectAsyncResult.cause().getMessage());
          }
        });

      });