在分布式系统里，SQL任务可能因为网络波动、数据库临时不可用、业务校验失败等原因执行失败，这时候需要设计可靠的重试记录机制，同时配合合适的方案保证数据最终一致，避免重试过程中出现数据重复、数据丢失等问题。
SQL任务重试记录机制设计
重试记录机制的核心是把每次SQL任务的执行状态、重试信息持久化存储，方便后续查询、重试调度和状态追踪，主要包含以下几个核心部分。
重试记录表设计
首先需要创建一张重试记录表，用来存储所有需要重试的SQL任务相关信息，表结构可以参考下面的设计：
CREATE TABLE sql_retry_record (
id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '主键ID',
task_id VARCHAR(64) NOT NULL COMMENT '业务任务唯一ID，保证幂等',
sql_content TEXT NOT NULL COMMENT '需要执行的SQL内容',
sql_params TEXT COMMENT 'SQL对应的参数，JSON格式存储',
status TINYINT NOT NULL DEFAULT 0 COMMENT '状态：0待执行 1执行中 2执行成功 3执行失败 4超过最大重试次数',
retry_count INT NOT NULL DEFAULT 0 COMMENT '当前重试次数',
max_retry_count INT NOT NULL DEFAULT 3 COMMENT '最大重试次数',
next_retry_time DATETIME COMMENT '下次重试时间',
error_msg TEXT COMMENT '最后一次执行失败的错误信息',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
UNIQUE KEY uk_task_id (task_id),
KEY idx_status_next_retry_time (status, next_retry_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='SQL任务重试记录表';
重试调度逻辑实现
重试调度需要定时扫描重试记录表中状态为待执行、执行失败，且下次重试时间小于当前时间的记录，然后执行对应的SQL任务，执行完成后更新记录状态。下面是一个简单的调度逻辑示例，使用Java实现：
import java.sql.*;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
public class SqlRetryScheduler {
// 数据库连接信息
private static final String JDBC_URL = "jdbc:mysql://127.0.0.1:3306/test_db?useUnicode=true&characterEncoding=utf8";
private static final String JDBC_USER = "root";
private static final String JDBC_PASSWORD = "123456";
// 扫描待重试的任务
public List<RetryRecord> scanPendingRetryTasks() {
List<RetryRecord> taskList = new ArrayList<>();
String sql = "SELECT id, task_id, sql_content, sql_params, retry_count, max_retry_count FROM sql_retry_record " +
"WHERE status IN (0, 3) AND next_retry_time <= ? LIMIT 100";
try (Connection conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PASSWORD);
PreparedStatement ps = conn.prepareStatement(sql)) {
ps.setTimestamp(1, Timestamp.valueOf(LocalDateTime.now()));
ResultSet rs = ps.executeQuery();
while (rs.next()) {
RetryRecord record = new RetryRecord();
record.setId(rs.getLong("id"));
record.setTaskId(rs.getString("task_id"));
record.setSqlContent(rs.getString("sql_content"));
record.setSqlParams(rs.getString("sql_params"));
record.setRetryCount(rs.getInt("retry_count"));
record.setMaxRetryCount(rs.getInt("max_retry_count"));
taskList.add(record);
}
} catch (SQLException e) {
e.printStackTrace();
}
return taskList;
}
// 执行SQL任务
public boolean executeSqlTask(RetryRecord record) {
String updateStatusSql = "UPDATE sql_retry_record SET status = ?, retry_count = ?, error_msg = ?, next_retry_time = ?, update_time = NOW() WHERE id = ?";
try (Connection conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PASSWORD)) {
conn.setAutoCommit(false);
// 执行目标SQL
try (PreparedStatement ps = conn.prepareStatement(record.getSqlContent())) {
// 这里可以解析sql_params设置参数，简化处理直接执行
ps.executeUpdate();
// 执行成功，更新状态为成功
try (PreparedStatement updatePs = conn.prepareStatement(updateStatusSql)) {
updatePs.setInt(1, 2);
updatePs.setInt(2, record.getRetryCount() + 1);
updatePs.setString(3, null);
updatePs.setTimestamp(4, null);
updatePs.setLong(5, record.getId());
updatePs.executeUpdate();
}
conn.commit();
return true;
} catch (SQLException e) {
// 执行失败，更新重试信息
conn.rollback();
int newRetryCount = record.getRetryCount() + 1;
int nextStatus = newRetryCount >= record.getMaxRetryCount() ? 4 : 3;
// 重试间隔采用指数退避策略，1s、2s、4s依次递增
LocalDateTime nextRetryTime = nextStatus == 3 ? LocalDateTime.now().plusSeconds((long) Math.pow(2, newRetryCount)) : null;
try (PreparedStatement updatePs = conn.prepareStatement(updateStatusSql)) {
updatePs.setInt(1, nextStatus);
updatePs.setInt(2, newRetryCount);
updatePs.setString(3, e.getMessage());
updatePs.setTimestamp(4, nextRetryTime == null ? null : Timestamp.valueOf(nextRetryTime));
updatePs.setLong(5, record.getId());
updatePs.executeUpdate();
}
conn.commit();
return false;
}
} catch (SQLException e) {
e.printStackTrace();
return false;
}
}
// 重试记录实体类
static class RetryRecord {
private Long id;
private String taskId;
private String sqlContent;
private String sqlParams;
private int retryCount;
private int maxRetryCount;
// 省略getter和setter
}
}
SQL保证最终一致的常见方案
重试机制只能解决临时失败的问题，要保证最终一致性还需要配合其他方案，下面介绍几种常用的实现方式。
幂等性设计
幂等性是指同一个SQL任务执行多次，产生的结果和执行一次相同，这是重试场景下保证数据一致的基础。实现幂等性的方式有很多，比如使用唯一业务ID，在SQL执行前先判断记录是否存在，或者使用数据库的唯一索引避免重复插入。
比如插入用户订单的场景，订单ID是唯一的，可以在订单表上对订单ID加唯一索引，重试插入时如果已经存在就会报错，捕获异常后判断是重复插入即可标记为执行成功：
-- 订单表，order_id加唯一索引
CREATE TABLE user_order (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
order_id VARCHAR(64) NOT NULL COMMENT '订单唯一ID',
user_id BIGINT NOT NULL COMMENT '用户ID',
order_amount DECIMAL(10,2) NOT NULL COMMENT '订单金额',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_order_id (order_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
事务补偿机制
如果SQL任务涉及多个数据库操作，或者跨服务调用，单个事务无法覆盖所有操作，这时候可以使用事务补偿机制。也就是执行正向操作的时候记录补偿SQL，如果后续操作失败，就执行补偿SQL回滚之前的操作。
比如转账场景，从A账户扣钱，给B账户加钱，两个操作需要保证一致，实现逻辑如下：
public boolean transferAmount(String taskId, Long fromUserId, Long toUserId, BigDecimal amount) {
// 先插入重试记录，sql_content存储正向操作SQL，sql_params存储参数
String insertRetrySql = "INSERT INTO sql_retry_record (task_id, sql_content, sql_params, status, max_retry_count, next_retry_time) VALUES (?, ?, ?, 0, 3, NOW())";
String forwardSql = "UPDATE user_account SET balance = balance - ? WHERE user_id = ? AND balance >= ?";
String compensateSql = "UPDATE user_account SET balance = balance + ? WHERE user_id = ?";
// 参数JSON格式：{"forwardParams":[amount,fromUserId,amount], "compensateParams":[amount,fromUserId], "toUserId":toUserId, "toAmount":amount}
try (Connection conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PASSWORD)) {
conn.setAutoCommit(false);
// 执行正向操作1：A账户扣钱
try (PreparedStatement ps = conn.prepareStatement(forwardSql)) {
ps.setBigDecimal(1, amount);
ps.setLong(2, fromUserId);
ps.setBigDecimal(3, amount);
int rows = ps.executeUpdate();
if (rows == 0) {
throw new RuntimeException("A账户余额不足");
}
}
// 执行正向操作2：B账户加钱
String addSql = "UPDATE user_account SET balance = balance + ? WHERE user_id = ?";
try (PreparedStatement ps = conn.prepareStatement(addSql)) {
ps.setBigDecimal(1, amount);
ps.setLong(2, toUserId);
ps.executeUpdate();
}
conn.commit();
return true;
} catch (Exception e) {
// 执行补偿操作，给A账户加回钱
try (Connection conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PASSWORD);
PreparedStatement ps = conn.prepareStatement(compensateSql)) {
ps.setBigDecimal(1, amount);
ps.setLong(2, fromUserId);
ps.executeUpdate();
} catch (SQLException ex) {
ex.printStackTrace();
}
// 记录重试任务，后续重试
return false;
}
}
本地消息表方案
本地消息表方案适合跨服务的SQL任务最终一致场景，核心思路是在本地数据库维护一张消息表，执行业务SQL和插入消息表放在同一个本地事务中，之后定时扫描消息表发送消息给其他服务，其他服务执行对应的SQL操作，执行完成后回调更新消息状态。
本地消息表结构可以参考：
CREATE TABLE local_message (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
message_id VARCHAR(64) NOT NULL COMMENT '消息唯一ID',
task_id VARCHAR(64) NOT NULL COMMENT '关联的业务任务ID',
message_content TEXT NOT NULL COMMENT '消息内容，包含需要执行的SQL信息',
status TINYINT NOT NULL DEFAULT 0 COMMENT '状态：0待发送 1已发送 2已消费 3消费失败',
retry_count INT NOT NULL DEFAULT 0 COMMENT '发送重试次数',
create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_message_id (message_id),
KEY idx_status_create_time (status, create_time)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='本地消息表';
业务操作和消息插入在同一个事务中，保证要么都成功要么都失败：
public boolean createOrderWithLocalMessage(String orderId, Long userId, BigDecimal amount) {
try (Connection conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PASSWORD)) {
conn.setAutoCommit(false);
// 1. 插入订单
String insertOrderSql = "INSERT INTO user_order (order_id, user_id, order_amount) VALUES (?, ?, ?)";
try (PreparedStatement ps = conn.prepareStatement(insertOrderSql)) {
ps.setString(1, orderId);
ps.setLong(2, userId);
ps.setBigDecimal(3, amount);
ps.executeUpdate();
}
// 2. 插入本地消息，通知库存服务扣减库存
String insertMessageSql = "INSERT INTO local_message (message_id, task_id, message_content, status) VALUES (?, ?, ?, 0)";
String messageId = "msg_" + orderId;
String messageContent = "{"sql":"UPDATE product_stock SET stock = stock - 1 WHERE product_id = 1001 AND stock >= 1", "params":[]}";
try (PreparedStatement ps = conn.prepareStatement(insertMessageSql)) {
ps.setString(1, messageId);
ps.setString(2, orderId);
ps.setString(3, messageContent);
ps.executeUpdate();
}
conn.commit();
return true;
} catch (SQLException e) {
e.printStackTrace();
return false;
}
}
注意事项
- 重试记录表的扫描频率不要太高，避免给数据库造成太大压力，同时扫描数量要加限制，防止一次性加载太多数据。
- 重试间隔建议采用指数退避策略，避免失败任务短时间大量重试加重系统负担。
- 最终一致方案的选择要根据实际业务场景，比如单库场景优先用幂等性+重试，跨服务场景可以考虑本地消息表或者事务补偿。
- 所有的重试和补偿操作都要记录详细的日志，方便后续排查问题。