[PHP] Seata AT 模式代码级详解

{
  "branchId":2828558179596595558,
  "sqlUndoLogs":[
    {
      "afterImage":{
        "rows":[
          {
            "fields":[
              {
                "keyType":"PRIMARY_KEY",
                "name":"id",
                "type":4,
                "value":3
              },
              {
                "keyType":"NULL",
                "name":"count",
                "type":4,
                "value":70
              }
            ]
          }
        ],
        "tableName":"stock_tbl"
      },
      "beforeImage":{
        "rows":[
          {
            "fields":[
              {
                "keyType":"PRIMARY_KEY",
                "name":"id",
                "type":4,
                "value":3
              },
              {
                "keyType":"NULL",
                "name":"count",
                "type":4,
                "value":100
              }
            ]
          }
        ],
        "tableName":"stock_tbl"
      },
      "sqlType":"UPDATE",
      "tableName":"stock_tbl"
    }
  ],
  "xid":"192.168.51.102:8091:2828558179596595550"
}

@Override
public void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) throws TransactionException {
Connection conn = null;b
ResultSet rs = null;
PreparedStatement selectPST = null;
boolean originalAutoCommit = true;
for (; ; ) {
try {
conn = dataSourceProxy.getPlainConnection();
// The entire undo process should run in a local transaction.
// 开启本地事务，确保删除undo log和恢复业务数据的SQL在一个事务中commit
if (originalAutoCommit = conn.getAutoCommit()) {
conn.setAutoCommit(false);
}
// Find UNDO LOG
selectPST = conn.prepareStatement(SELECT_UNDO_LOG_SQL);
selectPST.setLong(1, branchId);
selectPST.setString(2, xid);
// 查出branchId的所有undo log记录，用来恢复业务数据
rs = selectPST.executeQuery();
boolean exists = false;
while (rs.next()) {
exists = true;
// It is possible that the server repeatedly sends a rollback request to roll back
// the same branch transaction to multiple processes,
// ensuring that only the undo_log in the normal state is processed.
int state = rs.getInt(ClientTableColumnsName.UNDO_LOG_LOG_STATUS);
// 如果state=1，说明可以回滚；state=1说明不能回滚
if (!canUndo(state)) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("xid {} branch {}, ignore {} undo_log", xid, branchId, state);
}
return;
}
String contextString = rs.getString(ClientTableColumnsName.UNDO_LOG_CONTEXT);
Map<String, String> context = parseContext(contextString);
byte[] rollbackInfo = getRollbackInfo(rs);
String serializer = context == null ? null : context.get(UndoLogConstants.SERIALIZER_KEY);
// 根据serializer获取序列化工具类
UndoLogParser parser = serializer == null ? UndoLogParserFactory.getInstance()
: UndoLogParserFactory.getInstance(serializer);
// 反序列化undo log，得到业务记录修改前后的明文
BranchUndoLog branchUndoLog = parser.decode(rollbackInfo);
try {
// put serializer name to local
setCurrentSerializer(parser.getName());
List<SQLUndoLog> sqlUndoLogs = branchUndoLog.getSqlUndoLogs();
if (sqlUndoLogs.size() > 1) {
Collections.reverse(sqlUndoLogs);
}
for (SQLUndoLog sqlUndoLog : sqlUndoLogs) {
TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dataSourceProxy.getDbType()).getTableMeta(
conn, sqlUndoLog.getTableName(), dataSourceProxy.getResourceId());
sqlUndoLog.setTableMeta(tableMeta);
AbstractUndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(
dataSourceProxy.getDbType(), sqlUndoLog);
undoExecutor.executeOn(conn);
}
} finally {
// remove serializer name
removeCurrentSerializer();
}
}
// If undo_log exists, it means that the branch transaction has completed the first phase,
// we can directly roll back and clean the undo_log
// Otherwise, it indicates that there is an exception in the branch transaction,
// causing undo_log not to be written to the database.
// For example, the business processing timeout, the global transaction is the initiator rolls back.
// To ensure data consistency, we can insert an undo_log with GlobalFinished state
// to prevent the local transaction of the first phase of other programs from being correctly submitted.
// See https://github.com/seata/seata/issues/489
if (exists) {
deleteUndoLog(xid, branchId, conn);
conn.commit();
if (LOGGER.isInfoEnabled()) {
LOGGER.info("xid {} branch {}, undo_log deleted with {}", xid, branchId,
State.GlobalFinished.name());
}
} else {
// 如果不存在undo log，可能是因为分支事务还未执行完成（比如，分支事务执行超时），TM发起了回滚全局事务的请求。
// 这个时候，往undo_log表插入一条记录，可以使分支事务提交的时候失败（undo log）
insertUndoLogWithGlobalFinished(xid, branchId, UndoLogParserFactory.getInstance(), conn);
conn.commit();
if (LOGGER.isInfoEnabled()) {
LOGGER.info("xid {} branch {}, undo_log added with {}", xid, branchId,
State.GlobalFinished.name());
}
}
return;
} catch (SQLIntegrityConstraintViolationException e) {
// Possible undo_log has been inserted into the database by other processes, retrying rollback undo_log
if (LOGGER.isInfoEnabled()) {
LOGGER.info("xid {} branch {}, undo_log inserted, retry rollback", xid, branchId);
}
} catch (Throwable e) {
if (conn != null) {
try {
conn.rollback();
} catch (SQLException rollbackEx) {
LOGGER.warn("Failed to close JDBC resource while undo ... ", rollbackEx);
}
}
throw new BranchTransactionException(BranchRollbackFailed_Retriable, String
.format("Branch session rollback failed and try again later xid = %s branchId = %s %s", xid,
branchId, e.getMessage()), e);
} finally {
try {
if (rs != null) {
rs.close();
}
if (selectPST != null) {
selectPST.close();
}
if (conn != null) {
if (originalAutoCommit) {
conn.setAutoCommit(true);
}
conn.close();
}
} catch (SQLException closeEx) {
LOGGER.warn("Failed to close JDBC resource while undo ... ", closeEx);
}
}
}
}

public interface Compressor {
  /**
   * compress byte[] to byte[].
   * @param bytes the bytes
   * @return the byte[]
   */
  byte[] compress(byte[] bytes);
  /**
   * decompress byte[] to byte[].
   * @param bytes the bytes
   * @return the byte[]
   */
  byte[] decompress(byte[] bytes);
}

public interface UndoLogParser {
  /**
   * Get the name of parser;
   * 
   * @return the name of parser
   */
  String getName();
  /**
   * Get default context of this parser
   * 
   * @return the default content if undo log is empty
   */
  byte[] getDefaultContent();
  /**
   * Encode branch undo log to byte array.
   *
   * @param branchUndoLog the branch undo log
   * @return the byte array
   */
  byte[] encode(BranchUndoLog branchUndoLog);
  /**
   * Decode byte array to branch undo log.
   *
   * @param bytes the byte array
   * @return the branch undo log
   */
  BranchUndoLog decode(byte[] bytes);
}

​public interface Executor<T> {​  /**   * Execute t.   *   * @param args the args   * @return the t   * @throws Throwable the throwable   */  T execute(Object... args) throws Throwable;}

public interface Executor<T> {
  /**
   * Execute t.
   *
   * @param args the args
   * @return the t
   * @throws Throwable the throwable
   */
  T execute(Object... args) throws Throwable;
}