不止于 MySQL：多数据库兼容的调度策略封装
在之前的讨论中，我们基于 SELECT FOR UPDATE SKIP LOCKED 构建了一个高性能的分布式调度引擎。但现实世界是复杂的：你的应用可能部署在云上（云数据库五花八门），也可能需要适配客户的私有环境（可能是 SQL Server，或是古老的 MySQL 5.7）。如果调度框架与特定数据库深度绑定，当数据库迁移或环境变化时，你将面临巨大的改造成本。
本文将介绍如何通过策略模式封装不同的锁机制，让调度内核能够无缝适配多种数据库。我们将深入探讨：
- 为什么需要多数据库兼容
- 三种核心策略的实现：
SkipLockedStrategy、ReadPastStrategy、OptimisticLockStrategy - 如何根据数据库元数据动态选择策略
- 边界情况处理：隔离级别差异、分页语法差异等
一、为什么要兼容多数据库？
1.1 云环境的多数据库选择
在云原生时代，企业可能同时使用多个云厂商的数据库服务：
- AWS：Aurora（MySQL/PostgreSQL 兼容）、RDS for SQL Server
- 阿里云：PolarDB（MySQL 兼容）、RDS for PostgreSQL、RDS for SQL Server
- 腾讯云：TDSQL（MySQL 兼容）、PostgreSQL 版
如果调度框架只支持 MySQL 8.0+ 的 SKIP LOCKED，那么在面对 Aurora MySQL 5.7（不支持 SKIP LOCKED）时就会陷入困境。
1.2 混合部署与数据库迁移
许多公司存在混合部署场景：部分业务在 IDC 使用 Oracle，部分在云上使用 PostgreSQL。当业务整合时，需要一套调度框架能同时对接这两种数据库。
数据库迁移也是常见需求：从 SQL Server 迁移到 PostgreSQL，或从 MySQL 5.7 升级到 MySQL 8.0。如果调度框架对数据库有强依赖，迁移期间可能需要重写大量代码。
1.3 客户定制与 SaaS 多租户
如果你开发的是商业软件或 SaaS 服务，客户可能要求部署在自己的数据库上。客户的环境可能是五花八门的：SQL Server 2012、MySQL 5.6、Oracle 11g……没有多数据库兼容能力，就会失去大量商机。
下图展示了多数据库兼容的必要性：
| 应用部署环境 | 数据库类型 |
|---|---|
| 云环境 A | Aurora MySQL 5.7 |
| 云环境 B | RDS PostgreSQL 13 |
| 客户 IDC | SQL Server 2016 |
| 混合云 | Oracle 12c + MySQL 8.0 |
| 调度内核 | 需兼容多数据库 |
|---|---|
| 锁策略选择器 | SkipLockedStrategy / ReadPastStrategy / OptimisticLockStrategy |
二、策略模式封装
策略模式的核心是定义一个统一的接口，每种数据库的锁机制对应一个实现类。调度内核只依赖接口，不关心具体实现，从而实现数据库无关性。
2.1 锁策略接口
public interface LockStrategy {
/**
* 领取任务
* @param limit 最多领取数量
* @return 领取到的任务列表
*/
List<Task> fetchTasks(int limit);
/**
* 获取策略支持的数据库类型（用于调试和日志）
*/
String getSupportedDatabase();
}
2.2 SkipLockedStrategy：标准 SKIP LOCKED 实现
适用于支持 SELECT FOR UPDATE SKIP LOCKED 的数据库：
- PostgreSQL 9.5+
- MySQL 8.0.1+ (InnoDB)
- Oracle 12c+
- MariaDB 10.6+ (部分支持)
@Component
@ConditionalOnProperty(name = "db.lock.strategy", havingValue = "skip-locked", matchIfMissing = false)
public class SkipLockedStrategy implements LockStrategy {
@Autowired
private JdbcTemplate jdbcTemplate;
@Override
public List<Task> fetchTasks(int limit) {
String sql = """
SELECT id, task_type, payload, status, execute_time, retry_times, max_retries
FROM task
WHERE status IN (0, 4) AND execute_time <= NOW()
ORDER BY priority DESC, execute_time ASC
LIMIT ? FOR UPDATE SKIP LOCKED
""";
List<Task> tasks = jdbcTemplate.query(sql, new TaskRowMapper(), limit);
if (!tasks.isEmpty()) {
claimTasks(tasks);
}
return tasks;
}
private void claimTasks(List<Task> tasks) {
// 批量更新状态为执行中
List<Long> ids = tasks.stream().map(Task::getId).collect(Collectors.toList());
String inClause = ids.stream().map(String::valueOf).collect(Collectors.joining(","));
String updateSql = String.format("UPDATE task SET status = 1, node_id = ?, update_time = NOW() WHERE id IN (%s)", inClause);
jdbcTemplate.update(updateSql, NodeIdHolder.getNodeId());
}
@Override
public String getSupportedDatabase() {
return ;
}
}
流程图：
DBScheduler -> DBScheduler -> SELECT ... FOR UPDATE SKIP LOCKED LIMIT ? -> 返回未锁定的行 -> UPDATE ... SET status=1 WHERE id IN (...) -> 提交任务到线程池
2.3 ReadPastStrategy：SQL Server 替代方案
SQL Server 不支持 SKIP LOCKED，但可以通过表提示 READPAST + UPDLOCK 实现类似效果。
@Component
@ConditionalOnProperty(name = "db.lock.strategy", havingValue = "readpast")
public class ReadPastStrategy implements LockStrategy {
@Autowired
private JdbcTemplate jdbcTemplate;
@Override
public List<Task> fetchTasks(int limit) {
// SQL Server 使用 TOP 和 WITH (UPDLOCK, READPAST, ROWLOCK)
String sql = """
SELECT TOP (?) id, task_type, payload, status, execute_time, retry_times, max_retries
FROM task WITH (UPDLOCK, READPAST, ROWLOCK)
WHERE status IN (0, 4) AND execute_time <= GETDATE()
ORDER BY priority DESC, execute_time ASC
""";
List<Task> tasks = jdbcTemplate.query(sql, new TaskRowMapper(), limit);
if (!tasks.isEmpty()) {
claimTasks(tasks);
}
return tasks;
}
private void claimTasks(List<Task> tasks) {
// SQL Server 的更新语法与 MySQL 类似
List<Long> ids = tasks.stream().map(Task::getId).collect(Collectors.toList());
String inClause = ids.stream().map(String::valueOf).collect(Collectors.joining(","));
String updateSql = String.format("UPDATE task SET status = 1, node_id = ?, update_time = GETDATE() WHERE id IN (%s)", inClause);
jdbcTemplate.update(updateSql, NodeIdHolder.getNodeId());
}
@Override
public String getSupportedDatabase {
;
}
}
说明：
UPDLOCK：对读取的行加更新锁，防止其他事务修改。READPAST：跳过已被其他事务锁定的行（相当于SKIP LOCKED）。ROWLOCK：建议使用行级锁，避免锁升级。
注意：SQL Server 的 READPAST 只能用于已提交读（Read Committed）隔离级别，且不会跳过已由当前事务锁定的行。
2.4 OptimisticLockStrategy：乐观锁方案
对于不支持行级锁跳过机制的数据库（如 MySQL 5.7、Oracle 11g、SQLite），可以使用乐观锁 + 重试。
@Component
@ConditionalOnProperty(name = "db.lock.strategy", havingValue = "optimistic")
public class OptimisticLockStrategy implements LockStrategy {
@Autowired
private JdbcTemplate jdbcTemplate;
private static final int MAX_RETRY = 3;
@Override
public List<Task> fetchTasks(int limit) {
// 第一步：查询待处理任务（不加锁）
String selectSql = """
SELECT id, task_type, payload, status, execute_time, retry_times, max_retries, version
FROM task
WHERE status IN (0, 4) AND execute_time <= NOW()
ORDER BY priority DESC, execute_time ASC
LIMIT ?
""";
List<Task> tasks = jdbcTemplate.query(selectSql, new TaskRowMapper(), limit);
if (tasks.isEmpty()) {
return Collections.emptyList();
}
// 第二步：尝试通过乐观锁更新每个任务
List<Task> claimed = new ArrayList<>();
for (Task task : tasks) {
for (int i = 0; i < MAX_RETRY; i++) {
int updated = jdbcTemplate.update("""
UPDATE task SET status = 1, node_id = ?, version = version + 1, update_time = NOW()
WHERE id = ? AND status IN (0,4) AND version = ?
""", NodeIdHolder.getNodeId(), task.getId(), task.getVersion());
(updated > ) {
claimed.add(task);
;
} {
jdbcTemplate.queryForObject(, (), task.getId());
(latest == || latest.getStatus() != ) {
;
}
task.setVersion(latest.getVersion());
}
}
}
claimed;
}
String {
;
}
}
优缺点：
- 优点：兼容任何支持事务的数据库，无需特殊语法。
- 缺点：性能较差（多次查询），高并发下竞争激烈时领取成功率低。
三、动态适配数据库版本
有了多种策略，接下来需要根据当前连接的数据库自动选择合适的策略。我们可以通过 JDBC 的 DatabaseMetaData 获取数据库产品名称和版本号，然后决定实例化哪个策略。
3.1 策略适配器
@Component
public class LockStrategyAdapter {
@Autowired
private DataSource dataSource;
@Autowired
private ApplicationContext applicationContext;
private LockStrategy cachedStrategy;
@PostConstruct
public void init() {
this.cachedStrategy = determineStrategy();
}
public LockStrategy getStrategy() {
return cachedStrategy;
}
private LockStrategy determineStrategy() {
try (Connection conn = dataSource.getConnection()) {
DatabaseMetaData meta = conn.getMetaData();
String productName = meta.getDatabaseProductName().toLowerCase();
int majorVersion = meta.getDatabaseMajorVersion();
int minorVersion = meta.getDatabaseMinorVersion();
// 记录数据库信息
log.info("Detected database: {} version {}.{}", productName, majorVersion, minorVersion);
// 决策树
if (productName.contains("postgresql") && majorVersion >= 9) {
return applicationContext.getBean(SkipLockedStrategy.class);
} (productName.contains()) {
(majorVersion > || (majorVersion == && minorVersion >= )) {
applicationContext.getBean(SkipLockedStrategy.class);
} {
applicationContext.getBean(OptimisticLockStrategy.class);
}
} (productName.contains() && majorVersion >= ) {
applicationContext.getBean(SkipLockedStrategy.class);
} (productName.contains()) {
applicationContext.getBean(ReadPastStrategy.class);
} {
log.warn(, productName);
applicationContext.getBean(OptimisticLockStrategy.class);
}
} (SQLException e) {
(, e);
}
}
}
决策树图示：
PostgreSQL -> 是 -> SkipLockedStrategy MySQL -> 是 -> 版本 >= 8.0.1? -> 是 -> SkipLockedStrategy / 否 -> OptimisticLockStrategy Oracle -> 是 -> 版本 >= 12c? -> 是 -> SkipLockedStrategy SQL Server -> ReadPastStrategy 其他 -> OptimisticLockStrategy
3.2 在调度器中使用策略
调度器不再直接依赖具体策略，而是通过 LockStrategyAdapter 获取策略：
@Component
public class DistributedTaskScheduler {
@Autowired
private LockStrategyAdapter strategyAdapter;
// ... 其他依赖
private void pollLoop() {
// 计算可用槽位 ...
List<Task> tasks = strategyAdapter.getStrategy().fetchTasks(availableSlots);
// 提交任务 ...
}
}
3.3 配置覆盖
自动检测虽然方便，但有时需要手动指定策略（例如测试环境）。我们可以允许通过配置文件覆盖自动检测：
@Value("${scheduler.lock.strategy.override:}")
private String overrideStrategy;
private LockStrategy determineStrategy() {
if (StringUtils.hasText(overrideStrategy)) {
// 根据配置名称返回对应的 bean
return applicationContext.getBean(overrideStrategy, LockStrategy.class);
}
// 否则自动检测...
}
application.yml 示例：
scheduler:
lock:
strategy:
override: readPastStrategy # 强制使用 SQL Server 策略
四、兼容性测试与边界情况
即使策略封装得再好，不同数据库在细节上仍有差异，需要逐一处理。
4.1 事务隔离级别差异
- PostgreSQL：无此问题，任何隔离级别下
SKIP LOCKED都正常工作。 - SQL Server：
READPAST提示只能在已提交读（Read Committed）隔离级别下使用。如果使用可重复读或可序列化，READPAST会被忽略。
MySQL：在可重复读（RR）隔离级别下，SKIP LOCKED 可能因间隙锁而失效。因此使用 SkipLockedStrategy 时，建议将会话隔离级别设为读已提交（RC）。可以在获取连接后设置：
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
或者在 Spring 配置中指定：
spring:
datasource:
hikari:
connection-init-sql: SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED
4.2 分页语法差异
不同数据库的分页语法千差万别，而我们的策略中需要 LIMIT 或 TOP 子句。解决方法是使用 Spring 的 PagingQuery 或自定义 SQL 模板。一个更简洁的方法是使用 jOOQ 或 MyBatis 的方言适配，但为了轻量级，我们可以在策略实现中根据数据库类型动态生成 SQL。
例如，我们可以为每个策略定义自己的 SQL 模板，而不是在代码中硬编码。也可以使用 DatabaseMetaData 获取数据库名，然后在 SkipLockedStrategy 中判断：
if (databaseProductName.contains("mysql")) {
sql = "SELECT ... LIMIT ? FOR UPDATE SKIP LOCKED";
} else if (databaseProductName.contains("postgresql")) {
sql = "SELECT ... LIMIT ? FOR UPDATE SKIP LOCKED";
} else if (databaseProductName.contains("oracle")) {
sql = "SELECT ... FETCH FIRST ? ROWS ONLY FOR UPDATE SKIP LOCKED";
}
但更好的做法是每个策略只针对一种数据库，避免内部判断。所以上述 SkipLockedStrategy 实际上应该拆分为 PostgreSqlSkipLockedStrategy、MySql8SkipLockedStrategy 等，但考虑到它们 SQL 差异很小，可以统一用一套 SQL（以 MySQL 语法为主，PostgreSQL 兼容，Oracle 不兼容），但 Oracle 需要使用 FETCH FIRST。我们可以为 Oracle 单独实现一个子类，或者利用 Spring 的 JdbcTemplate 支持 ? 占位符的特性，但分页语法无法统一。
一种可行方案：使用 Spring Data JPA 或 MyBatis 的方言模块，或者干脆只支持主流数据库，忽略 Oracle。但若必须支持 Oracle，可以单独实现 OracleSkipLockedStrategy。
4.3 锁超时与死锁
不同数据库的锁超时默认值不同：
- MySQL：
innodb_lock_wait_timeout默认 50 秒。 - PostgreSQL：
lock_timeout默认 0（无限制）。 - SQL Server：
LOCK_TIMEOUT默认 -1（无限制）。
在使用乐观锁策略时，不会发生锁等待，因此无需关心。但在使用 SkipLockedStrategy 和 ReadPastStrategy 时，由于没有等待，锁超时实际上不会触发（因为压根不等待）。但要注意，如果 SKIP LOCKED 找不到足够行，事务可能很快提交，不会长时间持有锁。
4.4 批量更新语法差异
在 claimTasks 中，我们使用了 UPDATE ... WHERE id IN (...)。这种语法在大多数数据库中都支持，但要注意 SQL Server 对 IN 列表项数量有限制（最多 2100 个参数）。如果批量拉取数量较大（如超过 1000），可能需要分批更新。
我们可以将 claimTasks 改为循环更新，或者使用临时表。但为简化，可以限制 batchSize 不超过 500，避免超过参数限制。
4.5 事务边界控制
使用 SkipLockedStrategy 时，fetchTasks 方法通常需要在一个事务内执行，因为 SELECT FOR UPDATE 必须在事务中才能生效。在 Spring 中，可以在方法上添加 @Transactional 注解，但要确保事务传播级别正确。
对于 OptimisticLockStrategy，由于不依赖长事务，可以不在 fetchTasks 上开启事务，而是让每个更新操作独立事务。
4.6 单元测试与模拟数据库
为了测试多数据库兼容性，可以使用 Testcontainers 启动不同数据库的 Docker 容器，针对每个策略编写集成测试。
@Testcontainers
public class LockStrategyIntegrationTest {
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15");
@Container
static MySQLContainer<?> mysql = new MySQLContainer<>("mysql:8.0");
@Container
static MSSQLServerContainer<?> sqlserver = new MSSQLServerContainer<>("mcr.microsoft.com/mssql/server:2019-latest");
@Test
public void testPostgresStrategy() {
// 使用 postgres 数据源测试 SkipLockedStrategy
}
@Test
public void testMySql57Strategy() {
// 使用 mysql:5.7 测试 OptimisticLockStrategy
}
}
五、总结与展望
通过策略模式的巧妙运用，我们将分布式调度内核与具体数据库解耦，实现了一套代码，跑遍所有数据库的目标。本文介绍了三种核心策略：
| 策略 | 适用数据库 | 原理 | 优点 | 缺点 |
|---|---|---|---|---|
| SkipLockedStrategy | PostgreSQL 9.5+, MySQL 8.0.1+, Oracle 12c+ | SELECT FOR UPDATE SKIP LOCKED | 高性能、无阻塞 | 依赖数据库特性 |
| ReadPastStrategy | SQL Server | WITH (UPDLOCK, READPAST) | 无需等待 | SQL Server 专用 |
| OptimisticLockStrategy | 任何支持事务的数据库 | 乐观锁 + 重试 | 兼容性极广 | 性能较低，高并发下领取成功率低 |
我们还实现了动态适配器，根据数据库元数据自动选择策略，并提供了配置覆盖的能力。最后，讨论了兼容性测试中需要注意的边界情况，如隔离级别、分页语法、锁超时等。
当然，这还不是终点。未来可以扩展：
- 支持更多数据库：如 DB2、MariaDB、TiDB 等。
- 引入连接池与动态数据源：支持多数据源混合。
- 集成 Seata 等分布式事务：处理跨库任务。
- 基于策略的监控与告警：根据数据库类型展示不同的指标。
