如何在Scala Slick中使用sqlu批量执行多条Insert语句

首先，你的核心需求是把原来逐条执行的数据库操作改成一次性批量执行，而非循环调用单条处理。这里分两种思路实现，一种是复用你现有的单条操作逻辑合并执行，另一种是更高效的原生批量插入优化。

思路1：复用现有单条逻辑，合并所有DBIO动作 #

步骤1：移除单条动作中的事务声明 #

你原来的executeStreamingAction里单独加了.transactionally，但如果要把所有操作放在同一个事务里执行，需要去掉这个单条的事务声明，改为在合并后的动作上统一添加：

override def executeStreamingAction(record: TFFactDataIngestionMessage, settings: GSConnectorSettings): DBIOAction[Int, NoStream, Effect] = {
  val sqlUAction = sqlu""" INSERT INTO ... """
  sqlUAction // 移除这里的.transactionally
}

步骤2：新增批量处理方法 #

替换原来的runDatabaseAction和逐条调用的逻辑，新增一个处理整个记录列表的方法：

// 替换原来的foreach调用
Await.result(runDatabaseBatchAction(typedRecords), getQueryTimeOut(settings))

// 新增批量执行方法
private def runDatabaseBatchAction(records: Seq[T]): Future[Seq[Int]] = {
  // 把每条记录转换成对应的DBIOAction
  val dbioActions = records.map(record => executeStreamingAction(record, settings))
  
  // 合并所有动作成一个单一的DBIOAction，DBIO.sequence会按顺序执行所有动作，返回每个动作的结果集合
  val combinedDbio = DBIO.sequence(dbioActions)
    .transactionally // 所有操作放在同一个事务中
    .withStatementParameters(statementInit = _.setQueryTimeout(getQueryTimeOut(settings).toSeconds.toInt))
  
  // 执行合并后的动作
  db.get.run(combinedDbio)
}

这样所有记录的插入操作会被放在一个事务里一次性执行，而非逐条单独事务。如果确实需要每条记录单独事务（不推荐，性能较差），可以把.transactionally放回executeStreamingAction里，但合并后的动作仍会逐条执行，每条都在独立事务中。

思路2：原生批量插入（推荐，性能更优） #

上面的方法本质还是执行多条单条插入语句，而真正的批量插入是生成一条包含所有数据的INSERT语句，性能会提升很多。如果你的场景是插入操作，可以直接用Slick的批量插入API：

步骤1：定义表映射与转换方法 #

假设你的表对应的Slick TableQuery是yourTable，先编写方法把每条记录转换成表对应的行类型：

// 假设你的表行类型是YourTableRow，编写转换方法
private def convertToTableRow(record: TFFactDataIngestionMessage): YourTableRow = {
  YourTableRow(
    // 填充record对应的字段
    field1 = record.field1,
    field2 = record.field2,
    // ...其他字段
  )
}

步骤2：实现批量插入动作 #

// 替换原来的foreach调用
Await.result(runBulkInsertAction(typedRecords), getQueryTimeOut(settings))

// 原生批量插入方法
private def runBulkInsertAction(records: Seq[TFFactDataIngestionMessage]): Future[Int] = {
  val rows = records.map(convertToTableRow)
  
  val bulkInsertDbio = yourTable ++= rows
    .transactionally
    .withStatementParameters(statementInit = _.setQueryTimeout(getQueryTimeOut(settings).toSeconds.toInt))
  
  db.get.run(bulkInsertDbio)
}

这种方式Slick会生成类似INSERT INTO table (col1, col2) VALUES (val1, val2), (val3, val4), ...的批量插入语句，数据库只需要处理一次请求，性能比逐条执行好很多。

注意事项 #

• 事务边界：如果希望所有插入要么全部成功要么全部失败，一定要在合并后的动作上添加.transactionally，避免出现部分插入成功的情况。
• 超时设置：批量操作的超时时间要足够长，因为处理大量数据比单条耗时更久。
• Await的使用：如果你的代码运行在异步环境（比如Play框架），尽量避免使用Await.result，而是用Future的map/flatMap处理异步结果，避免阻塞线程。

内容的提问来源于stack exchange，提问作者skjagini