Quartz job分布式如何保证只调用一次的？

### Quartz 分布式如何保证只调用一次 在分布式环境中，Quartz 通过 locks 表来保证同一个 Job 只会被一个节点执行。Quartz 支持分布式，多个节点可能会同时抢占相同的资源，Quartz 依赖 locks 表来处理这种情况。以下是 Quartz 框架的一些基础概念和关键表： #### Quartz 框架的基础概念： - **Job 触发（fire）**：Quartz 把触发 job 叫做 fire。 - **TRIGGER_STATE**：当前 trigger 的状态。 - **PREV_FIRE_TIME**：上一次触发时间。 - **NEXT_FIRE_TIME**：下一次触发时间。 - **misfire**：指 job 在某一时刻要触发，却因为某些原因没有触发的情况。 #### Quartz 关键表： 1. **triggers 表**：记录某个 trigger 的 `PREV_FIRE_TIME`、`NEXT_FIRE_TIME`、`TRIGGER_STATE` 等信息。 2. **locks 表**：用于处理多个线程同时抢占相同资源的情况。 3. **fired_triggers 表**：记录正在触发的 trigger 的信息。 #### TRIGGER_STATE 状态： - **WAITING**：等待触发 - **ACQUIRED**：被调度线程拉取到 - **EXECUTING**：正在执行 - **COMPLETE**：执行完成 - **ERROR**：执行出错 - **PAUSED**：手动暂停 ### 分布式状态下的数据访问 Quartz 支持分布式，如果起了多个 Quartz 服务，会有多个调度线程来抢夺触发同一个 trigger。为了避免一个 trigger 被多个线程同时执行，Quartz 使用了 locks 表来保证同一时刻只有一个线程可以执行某个 trigger。 #### executeInNonManagedTXLock 方法 首先，来看 JobStoreSupport 类的 `executeInNonManagedTXLock()` 方法： ```java /** * Execute the given callback having acquired the given lock. * Depending on the JobStore, the surrounding transaction may be * assumed to be already present (managed). * * @param lockName The name of the lock to acquire, for example * "TRIGGER_ACCESS". If null, then no lock is acquired, but the * lockCallback is still executed in a transaction. */ protected Object executeInNonManagedTXLock(String lockName, TransactionCallback txCallback) throws JobPersistenceException { if (lockName == null) { return executeInNonManagedTXLock(txCallback); } Connection conn = null; try { conn = getNonManagedTXConnection(); if (conn == null) { throw new JobPersistenceException("Couldn't get non-managed TX connection."); } acquireLock(conn, lockName); return txCallback.execute(conn); } catch (JobPersistenceException e) { throw e; } finally { releaseLock(conn, lockName); cleanupConnection(conn); } } ``` 该方法通过获取指定的锁来执行传入的回调方法，并确保事务的一致性。 #### obtainLock 方法 接下来，来看一下 `obtainLock` 方法的实现： ```java public void obtainLock(Connection conn, String lockName) throws SQLException { if (!isLockOwner(conn, lockName)) { expandedSQL = "SELECT * FROM " + getTablePrefix() + "LOCKS WHERE LOCK_NAME = ? FOR UPDATE"; expandedInsertSQL = "INSERT INTO " + getTablePrefix() + "LOCKS (LOCK_NAME) VALUES (?)"; try (PreparedStatement ps = conn.prepareStatement(expandedSQL)) { ps.setString(1, lockName); ResultSet rs = ps.executeQuery(); if (!rs.next()) { try (PreparedStatement ps2 = conn.prepareStatement(expandedInsertSQL)) { ps2.setString(1, lockName); ps2.executeUpdate(); } } } } locks.put(lockName, Boolean.TRUE); } ``` 该方法通过锁住 locks 表的一行（由 lockName 确定）来确保线程安全。拿到锁后，会将 lockName 写入 threadlocal，在 `releaseLock` 时将其删除。 #### 执行流程 1. **调度线程扫描 triggers 表**：根据 `NEXT_FIRE_TIME` 提前拉取即将触发的 trigger。 2. **修改状态为 ACQUIRED**：如果被调度线程拉取到，状态变为 ACQUIRED。 3. **修改状态为 EXECUTING**：到达真正触发时刻，状态变为 EXECUTING。 4. **完成后修改状态**：如果不再执行，状态改为 COMPLETE；否则改为 WAITING，开始新的周期。若有异常，状态改为 ERROR；若手动暂停，状态改为 PAUSED。 #### 保障机制 通过 `executeInNonManagedTXLock()` 方法，Quartz 保证在分布式情况下，同一时刻只有一个线程可以执行某个方法，从而确保同一个 job 只会被一个节点执行。 ### 总结 通过 locks 表和 `executeInNonManagedTXLock()` 方法，Quartz 在分布式环境下保证了同一时刻只有一个线程可以执行某个 trigger，确保了 job 的唯一执行。