开源分布式数据库PolarDB-X源码解读——PolarDB-X源码解读（十一）：事务的一生

CREATE TABLE `tb1` (
    `id` int PRIMARY KEY,
    `a` int
) DBPARTITION BY HASH(`id`)

INSERT INTO tb1 VALUES (0, 0), (1, 1), (2, 2), (3, 3);

-- Transaction 1: 
BEGIN; 
SELECT * FROM tb1 WHERE id = 0; 
UPDATE tb1 SET a = 100 WHERE id = 1;
COMMIT;
-- Transaction 2: 
BEGIN; 
SELECT * FROM tb1 WHERE id = 0; 
UPDATE tb1 SET a = 101 WHERE id = 1;
UPDATE tb1 SET a = 101 WHERE id = 0;
COMMIT;

// TConnection#executeSQL(ByteString, Parameters, TStatement, ExecutionContext)
public ResultSet executeSQL(ByteString sql, Parameters params, TStatement stmt,
                                ExecutionContext executionContext) throws SQLException {
  if (this.trx == null || this.trx.isClosed()) {
    // 开启事务后，直到执行第一条语句，才会创建事务对象。
    beginTransaction();
  }
  // 让 executionContext 引用 trx 对象，方便后续执行器通过 trx 对象拿物理连接。
  executionContext.setTransaction(this.trx);
  resultCursor = executeQuery(sql, executionContext, trxPolicyModified);
}
// TConnection#beginTransaction(boolean)
private void beginTransaction(boolean autoCommit) {
  // 根据一些默认的或用户设定的事务变量，选择合适的事务策略，比如 TSO/XA 等。
  trxPolicy = loadTrxPolicy(executionContext);
  TransactionClass trxConfig = trxPolicy.getTransactionType(autoCommit, readOnly);
  // 根据事务策略，创建出对应的事务对象。
  this.trx = transactionManager.createTransaction(trxConfig, executionContext);
}
在我们的例子中，如果在上述代码打个断点，可以看到创建出来的是TsoTransaction，其中一些值得关注的变量为：
trx = {TsoTransaction}
    // 此时还没有获取任何时间戳。
    snapshotTimestamp = -1
    commitTimestamp = -1
    // 事务写的第一个物理分片（下称主分片），事务日志将会写在这个分片上。
  primaryGroup = null
    // 该事务是否跨分片，如果是单分片事务，会优化为一阶段提交。
    isCrossGroup = false
    // 事务日志管理器，负责写事务日志。
    globalTxLogManager = {GlobalTxLogManager}
    // 分布式事务的 connectionHolder 都是 TransactionConnectionHolder，
    // 里面分别存储了读写连接，读连接和写连接在提交时行为会有所不同。
    connectionHolder = {TransactionConnectionHolder}
        // 物理分片到对应写连接的映射。
        groupHeldWriteConn = {HashMap}
        // 物理分片到对应读连接集合到映射。在 ShareReadView 优化开启时，
        // 可以同时存在多个读连接和一个写连接，因此这里读写连接需要分开管理。
        // 该优化不在本文展开。
        groupHeldReadConns = {HashMap}

// AbstractTransaction#getConnection(String, String, IDataSource, RW, ExecutionContext)
public IConnection getConnection(String schema, String group, IDataSource ds, RW rw, ExecutionContext ec) {
  if (/* 是事务的第一个写请求 */) {
    // 把这个分片作为主分片，事务日志将写在这个分片上。
    this.primaryGroup = group;
    // 该分片还用于生成 XA 事务的 xid。
    this.primaryGroupUid = IServerConfigManager.getGroupUniqueId(schema, group);
  }
  // 通过 connectionHolder 拿到物理连接。
  IConnection conn = connectionHolder.getConnection(schema, group, ds, rw);
  if (/* 是写请求 */ && !isCrossGroup && !this.primaryGroup.equals(group)) {
    // 事务涉及了多个分片。
    this.isCrossGroup = true;
  }
  return conn;
}

// TransactionConnectionHolder#getConnection(String, String, IDataSource, RW)
public IConnection getConnection(String schema, String group, IDataSource ds, RW rw) {
  // 尝试获取该分片上的写连接，如果有，直接返回这个写连接。
  HeldConnection groupWriteConn = groupHeldWriteConn.get(group);
  if (groupWriteConn != null) {
    return groupWriteConn.connection;
  }
  HeldConnection freeReadConn = /* 尝试找到读连接 */;
  if (freeReadConn != null) {
    if (/* 当前需要写连接 */) {
      // 设置当前连接为写连接，participated = true 意味着该连接是写连接。
      freeReadConn.participated = true;
      this.groupHeldWriteConn.put(group, freeReadConn);
      // 由于原本是读连接，这里才真正开启分布式事务。
      this.trx.commitNonParticipant(group, freeReadConn.connection);
      this.trx.begin(schema, freeReadConn.group, freeReadConn.connection);
    }
    return freeReadConn.connection;
  }
  // 当前分片没有任何连接，创建一个新的连接。还会根据读写类型，
  // 设置好写连接 groupHeldWriteConn 或读连接集合 groupHeldReadConns。
  IConnection conn = new DeferredConnection(/* 这里会获取并封装该分片的私有协议连接 */);
  if (/* 需要写连接 */) {
    // 开启正常的分布式事务。
    this.trx.begin(schema, group, conn);
  } else {
    // 优化为只读事务。
    this.trx.beginNonParticipant(group, conn);
  }
  return conn;
}

// TsoTransaction#beginNonParticipant(String, IConnection)
protected void beginNonParticipant(String group, IConnection conn) throws SQLException {
  if (snapshotTimestamp < 0) {
    // 该事务从未拿过时间戳，则在这里获取。
    snapshotTimestamp = nextTimestamp();
  }
  // 使用私有协议的流水线执行机制执行 BEGIN。
  conn.executeLater("BEGIN");
  // 在 BEGIN 后发送时间戳。
  sendSnapshotSeq(conn);
}

// TsoTransaction#begin(String, String, IConnection)
protected void begin(String schema, String group, IConnection conn) throws SQLException {
  if (snapshotTimestamp < 0) {
    // 该事务从未拿过时间戳，则在这里获取。
    snapshotTimestamp = nextTimestamp();
  }
  // 获取 xid。
  String xid = getXid(group);
  // 触发私有协议流水线执行 XA START。
  conn.executeLater("XA START " + xid);
  // 在 XA START 后发送时间戳。
  sendSnapshotSeq(conn);
}

// TConnection#commit()
public void commit() throws SQLException {
  try {
    // 触发事务的提交流程。
    this.trx.commit();
  } finally {
    // 大部分情况下，如果事务提交成功，或出现异常后正确处理了，
    // 所有连接会被关闭并释放，trx.close() 相当于什么也不做。
    // 但如果事务提交失败且没有处理，这里会回滚事务并释放所有物理连接。
    this.trx.close();
    // 去掉对这个事务对象的引用，意味着当前连接里这个事务一生的结束。
    this.trx = null;
  }
}

// ShareReadViewTransaction#commit()
public void commit() {
  if (!isCrossGroup) {
    // 如果只涉及了单个分片的写，进行一阶段提交优化。
    commitOneShardTrx();
  } else {
    // 正常的多分片分布式事务提交流程。
    commitMultiShardTrx();
  }
}

// ShareReadViewTransaction#commitOneShardTrx()
protected void commitOneShardTrx() {
  // 对持有的所有物理连接执行以下流程，以提交每个物理连接开启的事务。
  forEachHeldConnection((group, conn, participated) -> {
      if (!participated) {
        // 只读连接是 BEGIN 开启事务的，且只有读操作，执行 ROLLBACK 即可。
        conn.execute("ROLLBACK");
      } else {
        // 获取 xid。
        String xid = getXid(group);
        // 写连接是 XA START 开启事务的，执行 XA END 和 XA COMMIT ONE PHASE 提交事务。
        conn.execute("XA END " + xid + "; XA COMMIT " + xid + " ONE PHASE");
      }
  });
  // 所有连接都提交了分支事务，释放并清空这些物理连接。
  connectionHolder.closeAllConnections();
}

// AbstractTransaction#close()
public void close() {
  // 回滚所有物理连接上的事务。
  cleanupAllConnections();
  // 释放并清空这些物理连接。
  connectionHolder.closeAllConnections();
}

// AbstractTransaction#cleanupAllConnections()
protected final void cleanupAllConnections() {
  // 对持有的所有物理连接执行以下流程，以回滚每个物理连接开启的事务。
  forEachHeldConnection((group, conn, participated) -> {
    if (conn.isClosed()) { return; }
    if (!participated) {
      // 只读连接是 BEGIN 开启事务的，且只有读操作，执行 ROLLBACK 即可。
      conn.execute("ROLLBACK");
    } else {
      // 获取 xid。
      String xid = getXid(group);
      // 写连接是 XA START 开启事务的，执行 XA END 和 XA ROLLBACK 回滚事务。
      conn.execute("XA END " + xid + "; XA ROLLBACK " + xid);
    }
  });
}

// TsoTransaction#commitMultiShardTrx()
protected void commitMultiShardTrx() {
  // 事务日志的提交状态，一共有 3 种：FAILURE，UNKNODWN，SUCCESS。
  TransactionCommitState commitState = TransactionCommitState.FAILURE;
  try {
    // 对所有连接执行 XA prepare。
    prepareConnections();
    // 拿 commit 时间戳。
    commitTimestamp = nextTimestamp();
    Connection logConn = /* 获取主分片上的连接 */;
    // 所有分支事务 prepare 成功，在写事务日志前，状态设置为 UNKNOWN，其作用见后续代码说明。
    commitState = TransactionCommitState.UNKNOWN;
    // 写事务日志。
    writeCommitLog(logConn);
    // 写事务日志成功，状态置为 SUCCESS。
    commitState = TransactionCommitState.SUCCESS;
  } catch (RuntimeException ex) {
    exception = ex;
  }
  if (commitState == TransactionCommitState.FAILURE) {
    // 写事务日志前失败了，回滚所有连接。
    rollbackConnections();
  } else if (commitState == TransactionCommitState.SUCCESS) {
    // 写事务日志成功了，提交所有连接。
    commitConnections();
  } else {
    // 所有连接都 prepare 成功了，但无法确定是否写入了事务日志，
    // 将由事务恢复的逻辑来决定是 COMMIT 还是 ROLLBACK。这里先丢弃所有连接。
    discardConnections();
  }
  // 释放并清空这些物理连接。
  connectionHolder.closeAllConnections();
  // prepare 或 commit 阶段抛出的异常，这里重新抛出。
  if (exception != null) {
      throw exception;
  }
}

// XARecoverTask#recoverInstance(IDataSource, Set<String>)
// dataSource 用户获取 DN 的连接，groups 是当前逻辑库的所有物理分片
private void recoverInstance(IDataSource dataSource, Set<String> groups) {
  // 执行 XA RECOVER 获取该 DN 上所有 prepare 的事务。
  // 此处省略了从 dataSource 获取连接和生成 statement 的代码。
  ResultSet rs = stmt.executeQuery("XA RECOVER");
  while (rs.next()) {
    // 对每一条记录，生成对象 PreparedXATrans，
    // 主要包括 xid，事务id，分支事务所在的分片，主分片等信息。
    PreparedXATrans trans = /* 从 rs 中获取一行数据生成 */;
    if (/* trans 所在分片是当前逻辑库的一个物理分片 && 
        trans 在上一次 recover 任务时也出现过，
        即较长时间都未被提交或回滚*/) {
      // 介入处理这个分支事务：回滚或提交。
      rollBackOrForward(trans, stmt);
    }
  }
}

// XARecoverTask#rollBackOrForward(PreparedXATrans, Statement)
private boolean rollBackOrForward(PreparedXATrans trans, Statement stmt) throws SQLException {
  String primaryGroup = /* 从 trans 里解析出主分片，详见前文关于 xid 的生成 */;
  // 尝试从主分片的事务日志中找到相关的事务日志。
  GlobalTxLog tx = GlobalTxLogManager.get(primaryGroup, trans.transId);
  if (tx != null) {
    // 确实存在事务日志，根据事务日志状态判断回滚或提交。
    if (tx.getState() == TransactionState.ABORTED) {
      // ABORTED 状态的事务需要回滚。
      return tryRollback(stmt, trans);
    } else {
      // SUCCESS 状态的事务需要提交。
      return tryCommitTSO(stmt, trans, tx.getCommitTimestamp());
    }
  } else {
    // 没有找到事务日志，尝试回滚，先开启事务，写下事务日志，标记事务为 ABORTED。
    try (Connection conn2 = /* 获取主分片上的另一个连接 */;) {
      conn2.setAutocommit(false);
      // 尝试回滚，如果因为别的线程正在处理这个事务
      //（比如该事务只是提交得慢，连接仍未断开，还在提交流程）
      // 而报错，就回滚上一条写事务日志的语句。
      txLog.append(transInfo.transId, TransactionType.XA, TransactionState.ABORTED, new ConnectionContext(), conn2);
      stmt.execute("XA ROLLBACK " + trans.toXid());
      conn2.commit();
      return true;
    } catch (Exception e) {
      /* 根据异常判断是否要回滚写事务日志的操作 */
    }
  }
}