在 Vertx (Vert.x 3) 中异步运行多个 Sql 查询

Posted 2023-04-19

【中文标题】在 Vertx (Vert.x 3) 中异步运行多个 Sql 查询

【英文标题】：Run Multiple Sql queries asynchronously in Vertx (Vert.x 3)

【发布时间】：2016-07-10 19:30:10

【问题描述】：

我正在使用 vertx-mysql-postgresql-client-3.3.0 和 vertx-sql-common-3.3.0 我最初想到使用批处理操作来插入多个语句，但看起来这个（http://vertx.io/docs/vertx-sql-common/java/#_batch_operations）在 vertx-mysql-postgres-client 中没有实现。

处理post数据到db的代码是：

SQLConnection sqlConnection;
 List <JsonArray> targetParams = new ArrayList();
 for (String key : targetObject.getTargets().keySet()) 
    targetParams.add(new JsonArray().add(key).add(targetObject.getTargets().get(key)).add(targetObject.getId()));
 
 int[] i = targetObject.getTargets().size();
 while(i[0]>=0) 
   sqlConnection.updateWithParams("INSERT INTO Targets (name,  id,language) VALUES (?,?,?)", taglineParams.get(i[0] - 1), result ->
           
  if (result.succeeded()) 
      i[0]--;
 

else  badRequest(context, "Error inserting error code: " +   result.cause().toString());
   
 ); 

AsyncResult 处理程序代码：

Handler<AsyncResult<SQLConnection>> sqlConnectionHandler = new Handler<AsyncResult<SQLConnection>>() 
    @Override
    public void handle(AsyncResult<SQLConnection> res) 
        if (res.succeeded()) 
            sqlConnection = res.result();
         else 
            logger.error("Unable to create SQL connection: " + res.cause().toString());
        
    
;

这个不起作用，先插入后出现线程阻塞错误。

知道如何使多个插入工作吗？

【参考方案1】：

Vert.x 是一个异步框架，但是代码使用的同步结构不起作用。在变量大于 0 之前有一个 while 循环。但是对控制变量的更新是在异步处理程序上更新的，这意味着它不会在循环内被调用。

知道这一点，循环将“永远”运行，直到 JVM 线程调度程序将停止它以运行其他线程（最终将更新控制变量）。现在调度程序已经停止了线程，Vert.x 也会通知您您阻止了事件循环，这意味着您应该修改您的代码。

我建议要么将 SQL 语句连接到大 sql 语句中并立即执行，要么使用 for 循环。请注意，for 循环也不是 100% 安全的。假设您正在插入 1M 行，您也可以轻松使用超过 2s 的 CPU 时间，您将再次收到警告。

较好的解决方案是异步循环或查看 RX。

【讨论】：

使用 for 循环我也得到了相同的线程已经运行的问题。像INSERT INTO Targets (name, id,language) VALUES (?,?,?),(?,?,?);", taglineParams, result -> 这样的查询会起作用吗？

【参考方案2】：

我强烈建议在这些情况下使用 CountDownLatch：

CountDownLatch latch = new CountDownLatch(targetObject.getTargets().size());
for (JsonArray params : targetParams) 
    sqlConnection.updateWithParams("INSERT INTO Targets (name,  id,language) VALUES (?,?,?)", params, result ->
    
        if (!result.succeeded()) 
            badRequest(context, "Error inserting error code: " +   result.cause().toString());
        
        latch.countDown();
    );


latch.await(3, TimeUnit.SECONDS);

这对 Vert.x 来说仍然很糟糕，因为它会阻塞线程，所以你还应该考虑改变你的逻辑，而不是等待进程完成。

【参考方案3】：

我看到了office API，batch operation，也许你可以用这个。

List<JsonArray> batch = new ArrayList<>();
batch.add(new JsonArray().add("joe"));
batch.add(new JsonArray().add("jane"));

connection.batchWithParams("INSERT INTO emp (name) VALUES (?)", batch, res -> 
  if (res.succeeded()) 
    List<Integer> result = res.result();
   else 
    // Failed!
  
);

【讨论】：

你测试过这个吗？当我检查代码时，它说操作不支持，因为它没有实现。