Прежде всего, поделитесь всеми предыдущими статьями, ставьте лайки, добавляйте в избранное и пересылайте три раза подряд. >>>>😜😜😜
Сборник статей:🎁nuggets.capable/post/694164…
Github :👉github.com/black-ant
Резервное копирование CASE:👉git ee.com/ant black/wipe…
Введение
Ранее упоминался процесс запроса Seata Client, в этой статье рассмотрим работу undo-log на стороне клиента.
undo-log является основной частью режима AT,Это делается в части RM, и данные undoLog будут генерироваться при обработке каждой единицы базы данных.
2. таблица журнала отмены
Давайте посмотрим на структуру таблицы undo-log.
CREATE TABLE `undo_log` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`branch_id` bigint(20) NOT NULL,
`xid` varchar(100) NOT NULL,
`context` varchar(128) NOT NULL,
`rollback_info` longblob NOT NULL,
`log_status` int(11) NOT NULL,
`log_created` datetime NOT NULL,
`log_modified` datetime NOT NULL,
`ext` varchar(100) DEFAULT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=17 DEFAULT CHARSET=utf8;
// 看一下其中可以了解的参数 :
- branch_id : 分支 ID
- context : 镜像数据
- rollback_info :
- log_status :
Оператор SQL
INSERT INTO `seata`.`undo_log`(
`id`, `branch_id`, `xid`,
`context`, `rollback_info`,
`log_status`, `log_created`, `log_modified`,
`ext`
) VALUES (
1, 5116237355214458898, '192.168.181.2:8091:5116237355214458897',
'serializer=jackson', 0x7B7D,
1, '2021-06-25 23:26:06', '2021-06-25 23:26:06',
NULL
);
Это не очень понятно, просто взглянув на оператор SQL, давайте рассмотрим его подробно,Здесь сначала раскрывается финальная логика обработки, при отладке можно использовать метод DEBUG для отката. :
// 看一下当前插入的 undoLog 详情
private void insertUndoLog(String xid, long branchId, String rollbackCtx,
byte[] undoLogContent, State state, Connection conn) throws SQLException {
try (PreparedStatement pst = conn.prepareStatement(INSERT_UNDO_LOG_SQL)) {
pst.setLong(1, 4386660905323926071);
pst.setString(2, "192.168.181.2:8091:4386660905323926065");
pst.setString(3, "serializer=jackson");
pst.setBlob(4, BlobUtils.bytes2Blob(undoLogContent));
pst.setInt(5, State.Normal(0));
pst.executeUpdate();
} catch (Exception e) {
if (!(e instanceof SQLException)) {
e = new SQLException(e);
}
throw (SQLException) e;
}
}
// undoLogContent 参数
{
"@class": "io.seata.rm.datasource.undo.BranchUndoLog",
"xid": "192.168.181.2:8091:4386660905323926065",
"branchId": 4386660905323926071,
"sqlUndoLogs": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.undo.SQLUndoLog",
"sqlType": "INSERT",
"tableName": "t_order",
"beforeImage": {
"@class": "io.seata.rm.datasource.sql.struct.TableRecords$EmptyTableRecords",
"tableName": "t_order",
"rows": ["java.util.ArrayList", []]
},
"afterImage": {
"@class": "io.seata.rm.datasource.sql.struct.TableRecords",
"tableName": "t_order",
"rows": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.sql.struct.Row",
"fields": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "id",
"keyType": "PRIMARY_KEY",
"type": 4,
"value": 31
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "order_no",
"keyType": "NULL",
"type": 12,
"value": "63098e74e93b49bba77f1957e8fdab39"
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "user_id",
"keyType": "NULL",
"type": 12,
"value": "1"
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "commodity_code",
"keyType": "NULL",
"type": 12,
"value": "C201901140001"
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "count",
"keyType": "NULL",
"type": 4,
"value": 50
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "amount",
"keyType": "NULL",
"type": 8,
"value": 100.0
}]]
}]]
}
}]]
}
undoLogContent — это преобразованный массив байтов BlobUtils.bytes2Blob, в котором хранятся xid и BranchId.Глобальный идентификатор транзакции (xid)так же какИдентификатор транзакции филиала (BranchId)и вошли в свойство sqlUndoLogsИмя таблицы (tableName)а такжеТип операции (sqlType)
пройти здесьbeforeImageа такжеafterImageДля данных до и после (PS: здесь бэкапится не вся запись, а бэкапятся некоторые параметры)
3. Поток обработки клиентского журнала отмены
Клиент предоставляет три реализации сохранения журнала отмены, как видите, все они сохраняются в библиотеку, но различается конкретный тип библиотеки.
3.1 Анализ AbstractUndoLogManager
AbstractUndoLogManager реализует UndoLogManager, который является основным инструментом управления и реализует управление журналом отмены.Этот класс в основном реализует следующие методы.
public interface UndoLogManager {
void flushUndoLogs(ConnectionProxy cp) throws SQLException;
void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) throws TransactionException;
void deleteUndoLog(String xid, long branchId, Connection conn) throws SQLException;
void batchDeleteUndoLog(Set<String> xids, Set<Long> branchIds, Connection conn) throws SQLException;
int deleteUndoLogByLogCreated(Date logCreated, int limitRows, Connection conn) throws SQLException;
}
Всего в кейсе три RM ( Order , Account , Storage ) Давайте посмотрим, как обрабатываются три RM
3.2 Процесс, инициированный журналом отмены (Заказ)
- ConnectionProxy # doCommit : Инициировать общий процесс фиксации
- ConnectionProxy # processGlobalTransactionCommit : Операция фиксации глобальной транзакции
- UndoLogManagerFactory # getUndoLogManager : Получить менеджер отмены журнала
- AbstractUndoLogManager # flushUndoLogs
- MySQLUndoLogManager # insertUndoLogWithNormal
- MySQLUndoLogManager # insertUndoLog : Вставить журнал отмены
3.2.1 Основной поток обработки журнала отмены
flushUndoLogs — это основной процесс.В этой ссылке запрашивается и создается BranchUndoLog.
public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
ConnectionContext connectionContext = cp.getContext();
if (!connectionContext.hasUndoLog()) {
return;
}
String xid = connectionContext.getXid();
long branchId = connectionContext.getBranchId();
// 构建undo-log 对象 -> 3.2.2 镜像的查询和获取
BranchUndoLog branchUndoLog = new BranchUndoLog();
branchUndoLog.setXid(xid);
branchUndoLog.setBranchId(branchId);
branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());
UndoLogParser parser = UndoLogParserFactory.getInstance();
byte[] undoLogContent = parser.encode(branchUndoLog);
// 插入数据 -> 3.2.3 最终数据的插入
insertUndoLogWithNormal(xid, branchId, buildContext(parser.getName()), undoLogContent,
cp.getTargetConnection());
}
// 这里 branchUndoLog 中存放了前后的数据 image , 可以来看一下
3.2.2 Запрос и получение изображений
Зеркалирование обрабатывает данные до изменения (beforeImage) и после изменения (AfterImage).Давайте посмотрим, где запрашивается зеркалирование.
// Image 的起点是 Context 中获取的
public class ConnectionProxy extends AbstractConnectionProxy {
private static final Logger LOGGER = LoggerFactory.getLogger(ConnectionProxy.class);
private ConnectionContext context = new ConnectionContext();
}
// 先来看一下 ConnectionContext 的结构 :
public class ConnectionContext {
private String xid;
private Long branchId;
private boolean isGlobalLockRequire;
/**
* Table and primary key should not be duplicated.
*/
private Set<String> lockKeysBuffer = new HashSet<>();
private List<SQLUndoLog> sqlUndoItemsBuffer = new ArrayList<>();
}
// 查询的流程 :
C- ExecuteTemplate # execute
C- AbstractDMLBaseExecutor # executeAutoCommitFalse
C- BaseTransactionalExecutor # prepareUndoLog
C- ConnectionContext # appendUndoItem
Step Start: основная логика, в которой запрашиваются переднее и заднее изображения.
// 在这个流程中 ,完成了大部分的数据操作
protected T executeAutoCommitFalse(Object[] args) throws Exception {
if (!JdbcConstants.MYSQL.equalsIgnoreCase(getDbType()) && isMultiPk()) {
throw new NotSupportYetException("multi pk only support mysql!");
}
// 查询前置镜像
TableRecords beforeImage = beforeImage();
// 执行 SQL 方法
T result = statementCallback.execute(statementProxy.getTargetStatement(), args);
// 查询后置镜像
TableRecords afterImage = afterImage(beforeImage);
// 保存 Undo-log
prepareUndoLog(beforeImage, afterImage);
return result;
}
// 补充 : TableRecords 对象
public class TableRecords implements java.io.Serializable {
// 支持序列化的能力
private static final long serialVersionUID = 4441667803166771721L;
private transient TableMeta tableMeta;
private String tableName;
private List<Row> rows = new ArrayList<Row>();
Шаг 1: Получить перед изображением
AbstractDMLBaseExecutor будет иметь несколько классов реализации в соответствии с различной обработкой.
Вот только пример Update, который не будет запрашиваться во время вставки, поэтому он не будет слишком углубляться:
// C-BaseInsertExecutor : Insert 情况时的处理方式
protected TableRecords beforeImage() throws SQLException {
return TableRecords.empty(getTableMeta());
}
// C-UpdateExecutor : Update 情况时 Image 的查询方式
protected TableRecords beforeImage() throws SQLException {
ArrayList<List<Object>> paramAppenderList = new ArrayList<>();
TableMeta tmeta = getTableMeta();
String selectSQL = buildBeforeImageSQL(tmeta, paramAppenderList);
return buildTableRecords(tmeta, selectSQL, paramAppenderList);
}
// Step 1-1 :
private String buildBeforeImageSQL(TableMeta tableMeta, ArrayList<List<Object>> paramAppenderList) {
SQLUpdateRecognizer recognizer = (SQLUpdateRecognizer) sqlRecognizer;
List<String> updateColumns = recognizer.getUpdateColumns();
assertContainsPKColumnName(updateColumns);
StringBuilder prefix = new StringBuilder("SELECT ");
StringBuilder suffix = new StringBuilder(" FROM ").append(getFromTableInSQL());
String whereCondition = buildWhereCondition(recognizer, paramAppenderList);
if (StringUtils.isNotBlank(whereCondition)) {
suffix.append(WHERE).append(whereCondition);
}
String orderBy = recognizer.getOrderBy();
if (StringUtils.isNotBlank(orderBy)) {
suffix.append(orderBy);
}
ParametersHolder parametersHolder = statementProxy instanceof ParametersHolder ? (ParametersHolder)statementProxy : null;
String limit = recognizer.getLimit(parametersHolder, paramAppenderList);
if (StringUtils.isNotBlank(limit)) {
suffix.append(limit);
}
suffix.append(" FOR UPDATE");
StringJoiner selectSQLJoin = new StringJoiner(", ", prefix.toString(), suffix.toString());
// 是否只更新列
if (ONLY_CARE_UPDATE_COLUMNS) {
if (!containsPK(updateColumns)) {
selectSQLJoin.add(getColumnNamesInSQL(tableMeta.getEscapePkNameList(getDbType())));
}
// 查询更新的列
for (String columnName : updateColumns) {
selectSQLJoin.add(columnName);
}
} else {
for (String columnName : tableMeta.getAllColumns().keySet()) {
selectSQLJoin.add(ColumnUtils.addEscape(columnName, getDbType()));
}
}
// SELECT id, count FROM t_storage WHERE commodity_code = ? FOR UPDATE
return selectSQLJoin.toString();
}
Метаданные таблицы TableMeta:
Шаг 2: запрос после изображения
protected TableRecords afterImage(TableRecords beforeImage) throws SQLException {
TableMeta tmeta = getTableMeta();
if (beforeImage == null || beforeImage.size() == 0) {
return TableRecords.empty(getTableMeta());
}
String selectSQL = buildAfterImageSQL(tmeta, beforeImage);
ResultSet rs = null;
try (PreparedStatement pst = statementProxy.getConnection().prepareStatement(selectSQL)) {
SqlGenerateUtils.setParamForPk(beforeImage.pkRows(), getTableMeta().getPrimaryKeyOnlyName(), pst);
rs = pst.executeQuery();
return TableRecords.buildRecords(tmeta, rs);
} finally {
IOUtil.close(rs);
}
}
// 这里就不深入了
Шаг 3: Добавьте журнал отмены, создайте Context()
C- BaseTransactionalExecutor
protected void prepareUndoLog(TableRecords beforeImage, TableRecords afterImage) throws SQLException {
// image 改变时才会创建 undo-log
if (beforeImage.getRows().isEmpty() && afterImage.getRows().isEmpty()) {
return;
}
// 获取代理连接器
ConnectionProxy connectionProxy = statementProxy.getConnectionProxy();
// 插入实体 -> 详见下图
TableRecords lockKeyRecords = sqlRecognizer.getSQLType() == SQLType.DELETE ? beforeImage : afterImage;
String lockKeys = buildLockKey(lockKeyRecords);
connectionProxy.appendLockKey(lockKeys);
SQLUndoLog sqlUndoLog = buildUndoItem(beforeImage, afterImage);
connectionProxy.appendUndoLog(sqlUndoLog);
}
После того, как изображение будет запрошено здесь, давайте посмотрим на процесс вставки изображения.
3.2.3 Вставка окончательных данных
protected void insertUndoLogWithGlobalFinished(String xid, long branchId, UndoLogParser parser, Connection conn) throws SQLException {
insertUndoLog(xid, branchId, buildContext(parser.getName()),
parser.getDefaultContent(), State.GlobalFinished, conn);
}
// 数据的插入逻辑在章节2中
Добавлено: зеркалирование журнала отмены (хранилище) при обновлении.
Основной процесс соответствует порядку.Давайте в основном посмотрим на данные журнала отмены при вставке.Вы можете видеть, что вместо генерации SQL поля и данные зеркалируются.
И зеркалирование здесь связано с изменением узлов
{
"@class": "io.seata.rm.datasource.undo.BranchUndoLog",
"xid": "192.168.181.2:8091:4386660905323926147",
"branchId": 4386660905323926150,
"sqlUndoLogs": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.undo.SQLUndoLog",
"sqlType": "UPDATE",
"tableName": "t_storage",
"beforeImage": {
"@class": "io.seata.rm.datasource.sql.struct.TableRecords",
"tableName": "t_storage",
"rows": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.sql.struct.Row",
"fields": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "id",
"keyType": "PRIMARY_KEY",
"type": 4,
"value": 1
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "count",
"keyType": "NULL",
"type": 4,
"value": -800
}]]
}]]
},
"afterImage": {
"@class": "io.seata.rm.datasource.sql.struct.TableRecords",
"tableName": "t_storage",
"rows": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.sql.struct.Row",
"fields": ["java.util.ArrayList", [{
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "id",
"keyType": "PRIMARY_KEY",
"type": 4,
"value": 1
}, {
"@class": "io.seata.rm.datasource.sql.struct.Field",
"name": "count",
"keyType": "NULL",
"type": 4,
"value": -850
}]]
}]]
}
}]]
}
AccountТочно так же я не буду говорить об этом здесь пока
4. Процесс отката клиента отмены журнала
Прочитав описанный выше процесс создания журнала отмены, давайте взглянем на обработку журнала отмены во время отката.
Тут есть очень важный момент знаний, создание undo-log создается в каждом RM, а вот откат -
4.1 процесс отката отмены журнала
Основной процесс отката:
- RmBranchRollbackProcessor # process : Получен запрос на резервную обработку
- RmBranchRollbackProcessor # handleBranchRollback
- AbstractRMHandler # onRequest
- AbstractRMHandler # handle
- AbstractExceptionHandler # exceptionHandleTemplate
- AbstractRMHandler # handle
- AbstractRMHandler # doBranchRollback : откат ветки
- DataSourceManager # branchRollback
- AbstractUndoLogManager # undo : Выполнить логику отмены
- AbstractUndoLogManager # deleteUndoLog : удалить ветку
Как вы можете видеть здесь, основная логика — это отмена, код этой логики относительно длинный, я делю его на две логики: обратный вызов и удаление журнала отмены:
4.2 Обратный вызов основной логики
C- AbstractUndoLogManager
public void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) throws TransactionException {
Connection conn = null;
ResultSet rs = null;
PreparedStatement selectPST = null;
boolean originalAutoCommit = true;
for (; ; ) {
conn = dataSourceProxy.getPlainConnection();
// The entire undo process should run in a local transaction.
if (originalAutoCommit = conn.getAutoCommit()) {
conn.setAutoCommit(false);
}
// 通过 branchId 和 xid 查询 undo-log
selectPST = conn.prepareStatement(SELECT_UNDO_LOG_SQL);
selectPST.setLong(1, branchId);
selectPST.setString(2, xid);
rs = selectPST.executeQuery();
boolean exists = false;
// 对查询出的 undo-log 进行循环处理
while (rs.next()) {
exists = true;
// 服务器可能会重复发送回滚请求，将同一个分支事务回滚到多个进程，从而确保只处理正常状态下的undo_log
int state = rs.getInt(ClientTableColumnsName.UNDO_LOG_LOG_STATUS);
if (!canUndo(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);
UndoLogParser parser = serializer == null ? UndoLogParserFactory.getInstance()
: UndoLogParserFactory.getInstance(serializer);
// 反序列化为 BranchUndoLog
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);
}
// 执行 undo-log 进行回退处理
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();
}
}
// ........ 省略回退逻辑
}
}
AbstractUndoExecutor откатывает executeOn
public void executeOn(Connection conn) throws SQLException {
if (IS_UNDO_DATA_VALIDATION_ENABLE && !dataValidationAndGoOn(conn)) {
return;
}
try {
// UPDATE t_storage SET count = ? WHERE id = ?
String undoSQL = buildUndoSQL();
PreparedStatement undoPST = conn.prepareStatement(undoSQL);
TableRecords undoRows = getUndoRows();
// 获取受影响的列
for (Row undoRow : undoRows.getRows()) {
ArrayList<Field> undoValues = new ArrayList<>();
List<Field> pkValueList = getOrderedPkList(undoRows, undoRow, getDbType(conn));
for (Field field : undoRow.getFields()) {
if (field.getKeyType() != KeyType.PRIMARY_KEY) {
undoValues.add(field);
}
}
// 解析需要回退的字段值 (即原有值)
undoPrepare(undoPST, undoValues, pkValueList);
// 执行 undo-log 处理 , 回退值
undoPST.executeUpdate();
}
} catch (Exception ex) {
}
}
5. Процесс удаления журнала отмены клиента
После того, как откат завершен, давайте посмотрим на обработку удаления undo-log.Логика удаления обрабатывается после логики отката.
5.1 Основная логика отмены регистрации
public void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) throws TransactionException {
Connection conn = null;
ResultSet rs = null;
PreparedStatement selectPST = null;
boolean originalAutoCommit = true;
for (; ; ) {
try {
// .... 省略 rollback 逻辑
// 如果undo_log存在，这意味着分支事务已经完成了第一阶段，我们可以直接回滚并清理undo_log
// 否则，它表明分支事务中有一个异常，导致undo_log没有写入数据库。
// 例如，业务处理超时时，全局事务被启动器回滚。
// 为了确保数据的一致性，我们可以插入一个带有GlobalFinished状态的undo_log，以防止其他程序第一阶段的本地事务被正确提交。
if (exists) {
deleteUndoLog(xid, branchId, conn);
conn.commit();
} else {
insertUndoLogWithGlobalFinished(xid, branchId, UndoLogParserFactory.getInstance(), conn);
conn.commit();
}
return;
} catch (SQLIntegrityConstraintViolationException e) {
// Possible undo_log has been inserted into the database by other processes, retrying rollback undo_log
} 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 {
//...
}
}
}
5.2 Удалить журнал отмены
public void deleteUndoLog(String xid, long branchId, Connection conn) throws SQLException {
try (PreparedStatement deletePST = conn.prepareStatement(DELETE_UNDO_LOG_SQL)) {
deletePST.setLong(1, branchId);
deletePST.setString(2, xid);
deletePST.executeUpdate();
} catch (Exception e) {
if (!(e instanceof SQLException)) {
e = new SQLException(e);
}
throw (SQLException) e;
}
}
Суммировать
Эта статья просто обобщает логику отмены журнала,В основном сохраняйте логику до и после с помощью BeforeImage и AfterImage., для резервной обработки
Но это далеко не конец,Есть также механизм блокировки и механизм удаленного вызова для улучшения всего процесса, и в то же время необходимо разобраться в логике TCC.