如何在存在外键关联的多表中实现并行插入数据？

我尝试在存在外键关联的多张表中并行插入数据，但触发了外键约束违例错误：

java.lang.Exception: org.postgresql.util.PSQLException: ERROR: insert or update on table "child2" violates foreign key constraint "child2_parent_id_fkey"

我知道当前写法有问题，但找不到合适的最佳实践。我的核心需求是：

• 尽可能实现并行插入
• 任意错误发生时，所有表的操作都能回滚

以下是我的Scala代码实现：

class Scratch (creds: DbCreds, dbType: String, driverClassName: String, initialPoolSize: Int = 4,
               maxPoolSize: Int = 10, queryTimeout: Int = 600000) extends DbConnectionPool with Configs {

  val connectionPool = setConnectionPool(creds, dbType, driverClassName, initialPoolSize, maxPoolSize, queryTimeout)
  implicit val ec = ExecutionContext.global

  def updateTables(): Unit = {

//     表结构定义：
//    create table parent (parent_id int primary key);
//    create table child (id int primary key, parent_id int references parent(parent_id));
//    create table child2 (id int primary key, parent_id int references parent(parent_id));

    val parentTableConn: Connection = getNonCommitConnection()
    val childTableConn: Connection  = getNonCommitConnection()
    val childTable2Conn: Connection = getNonCommitConnection()

    try {
      val parentTableFuture = Future {
        updateParent(parentTableConn)
      }
      parentTableFuture.onComplete {
        case Success(x) => println("-parent table updated successfully")
        case Failure(error) => throw new Exception(error)
      }
      Await.result(parentTableFuture, queryTimeout millisecond)
      val childTableFuture = Future {
        updateChild(childTableConn)
      }
      childTableFuture.onComplete {
        case Success(x) => println("-child table updated successfully")
        case Failure(error) => throw new Exception(error)
      }
      val childTable2Future = Future {
        updateChild2(childTable2Conn)
      }
      childTable2Future.onComplete {
        case Success(x) => println("-child2 table updated successfully")
        case Failure(error) => throw new Exception(error)
      }
      parentTableConn.commit()
      childTableConn.commit()
      childTable2Conn.commit()

    } catch {
      case e: Exception => {
        println(s"ERROR: exception thrown rolling back db commits
${e}")
        parentTableConn.rollback()
        childTableConn.rollback()
        childTable2Conn.rollback()

        throw e
      }
    } finally {
      parentTableConn.close()
      childTableConn.close()
      childTable2Conn.close()
    }
  }

  def updateParent(conn: Connection): Unit = {
    val sql = "insert into parent (parent_id) values (1),(2),(3);"
    val stmt = conn.createStatement()
    stmt.execute(sql)
    stmt.close()
  }

  def updateChild(conn: Connection): Unit = {
    val sql = "insert into child (id, parent_id) values (1,1),(2,2),(3,1);"
    val stmt = conn.createStatement()
    stmt.execute(sql)
    stmt.close()
  }

  def updateChild2(conn: Connection): Unit = {
    val sql = "insert into child2 (id, parent_id) values (1,3),(2,3),(3,1);"
    val stmt = conn.createStatement()
    stmt.execute(sql)
    stmt.close()
  }

}

问题分析 #

你的代码核心问题在于使用了三个独立的数据库连接：

1. 每个连接的事务相互隔离，parent表插入后未提交时，child/child2的连接看不到新插入的parent数据，直接触发外键约束检查失败
2. 最后分别提交三个连接的事务，无法保证原子性——如果其中一个提交失败，另外两个已提交的操作无法回滚

解决方案（兼顾并行与事务原子性） #

方案1：延迟外键约束 + 单连接并行执行（推荐） #

步骤1：修改外键为可延迟约束

PostgreSQL支持延迟外键，会在事务提交时才检查约束，而非每条语句执行时检查：

-- 修改child表外键
ALTER TABLE child 
DROP CONSTRAINT child_parent_id_fkey;
ALTER TABLE child 
ADD CONSTRAINT child_parent_id_fkey 
FOREIGN KEY (parent_id) REFERENCES parent(parent_id)
DEFERRABLE INITIALLY DEFERRED;

-- 修改child2表外键
ALTER TABLE child2 
DROP CONSTRAINT child2_parent_id_fkey;
ALTER TABLE child2 
ADD CONSTRAINT child2_parent_id_fkey 
FOREIGN KEY (parent_id) REFERENCES parent(parent_id)
DEFERRABLE INITIALLY DEFERRED;

步骤2：单连接内并行执行插入

所有操作复用同一个连接，在同一个事务内并行执行插入，既保证效率，又能实现原子回滚：

class Scratch (creds: DbCreds, dbType: String, driverClassName: String, initialPoolSize: Int = 4,
               maxPoolSize: Int = 10, queryTimeout: Int = 600000) extends DbConnectionPool with Configs {

  val connectionPool = setConnectionPool(creds, dbType, driverClassName, initialPoolSize, maxPoolSize, queryTimeout)
  implicit val ec = ExecutionContext.global

  def updateTables(): Unit = {
    // 表结构已修改为带延迟外键
    val conn: Connection = getNonCommitConnection()
    try {
      // 并行执行三个插入任务，共享同一个连接与事务
      val parentFuture = Future { updateParent(conn) }
      val childFuture = Future { updateChild(conn) }
      val child2Future = Future { updateChild2(conn) }

      // 等待所有并行任务完成
      Await.result(Future.sequence(List(parentFuture, childFuture, child2Future)), queryTimeout millisecond)
      
      // 提交事务，此时才统一检查外键约束
      conn.commit()
      println("所有表更新成功")
    } catch {
      case e: Exception => {
        println(s"ERROR: 执行失败，回滚事务\n${e}")
        conn.rollback()
        throw e
      }
    } finally {
      conn.close()
    }
  }

  def updateParent(conn: Connection): Unit = {
    val sql = "insert into parent (parent_id) values (1),(2),(3);"
    val stmt = conn.createStatement()
    stmt.execute(sql)
    stmt.close()
  }

  def updateChild(conn: Connection): Unit = {
    val sql = "insert into child (id, parent_id) values (1,1),(2,2),(3,1);"
    val stmt = conn.createStatement()
    stmt.execute(sql)
    stmt.close()
  }

  def updateChild2(conn: Connection): Unit = {
    val sql = "insert into child2 (id, parent_id) values (1,3),(2,3),(3,1);"
    val stmt = conn.createStatement()
    stmt.execute(sql)
    stmt.close()
  }
}

方案2：分布式XA事务（仅适用于多连接/分库场景） #

如果必须使用多个独立连接（如分库），可以通过XA事务协调多个连接的事务，保证原子性。但该方案复杂度高、性能开销大，PostgreSQL需额外配置（如启用max_prepared_transactions），非必要不推荐。

内容的提问来源于stack exchange，提问作者J.Hammond