XXL-JOB - 虾米的个人笔记

核心执行逻辑(版本2.2.0)#

com.xxl.job.admin.core.thread.JobScheduleHelper#start

scheduleThread#

该线程负责从数据库中预读将要执行的任务，并将其放入到 ringData 中用于后续真正的调度线程进行调度

1. 线程初始化

1
scheduleThread = new Thread(new Runnable() {
2
    @Override
3
    public void run() {
4
        // 线程执行的逻辑
5
    }
6
});

创建了一个新的线程 scheduleThread，并实现了 Runnable 接口的 run 方法。

2. 线程启动前的延迟

1
try {
2
    TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
3
} catch (InterruptedException e) {
4
    if (!scheduleThreadToStop) {
5
        logger.error(e.getMessage(), e);
6
    }
7
}

线程启动后，首先会休眠一段时间，确保线程在整秒时刻启动（即 System.currentTimeMillis() % 1000 为 0 的时刻）。这样可以使得后续的调度任务能够对齐到整秒。

3. 预读任务数量计算

1
int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;

计算预读任务的数量。preReadCount 是根据线程池的大小和触发器的 QPS（每秒查询率）来计算的。假设每个触发器耗时 50ms，那么 QPS 为 20（1000ms / 50ms）

4. 主循环

1
while (!scheduleThreadToStop) {
2
    // 主循环逻辑
3
}

主循环会一直运行，直到 scheduleThreadToStop 被设置为 true

5. 数据库连接与事务管理

1
Connection conn = null;
2
Boolean connAutoCommit = null;
3
PreparedStatement preparedStatement = null;
4

5
try {
6
    conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
7
    connAutoCommit = conn.getAutoCommit();
8
    conn.setAutoCommit(false);
9

10
    preparedStatement = conn.prepareStatement("select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
11
    preparedStatement.execute();

获取数据库连接，并开启事务。通过 for update 语句锁定 xxl_job_lock 表中的 schedule_lock 记录，确保在同一时间只有一个调度线程在执行
这里要思考如果获取到数据库悲观锁之后，JVM挂掉，会不会一直无法释放锁导致其他节点无法获取数据库悲观锁

6. 预读任务

1
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);

从数据库中预读即将触发的任务列表。nowTime + PRE_READ_MS 表示预读的时间范围，preReadCount 是预读的任务数量。

7. 任务处理

1
for (XxlJobInfo jobInfo: scheduleList) {
2
    if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
3
        // 任务过期处理
4
    } else if (nowTime > jobInfo.getTriggerNextTime()) {
5
        // 任务触发处理
6
    } else {
7
        // 任务预读处理
8
    }
9
}

遍历预读的任务列表，根据任务的触发时间进行不同的处理：
- 任务过期：如果任务的触发时间已经超过预读时间范围，则标记任务为过期，并更新下一次触发时间。
- 任务触发：如果任务的触发时间在当前时间和预读时间之间，则立即触发任务，并更新下一次触发时间。
- 任务预读：如果任务的触发时间在未来，则将任务放入时间环中，等待后续触发。

8. 更新任务信息

1
if (conn != null) {
2
    try {
3
        conn.commit();
4
    } catch (SQLException e) {
5
        if (!scheduleThreadToStop) {
6
            logger.error(e.getMessage(), e);
7
        }
8
    }
9
    // 恢复自动提交状态并关闭连接
10
}

更新数据库中任务的信息，确保任务的触发时间和状态是最新的。

9. 事务提交与资源释放

1
if (conn != null) {
2
    try {
3
        conn.commit();
4
    } catch (SQLException e) {
5
        if (!scheduleThreadToStop) {
6
            logger.error(e.getMessage(), e);
7
        }
8
    }
9
    // 恢复自动提交状态并关闭连接
10
}

提交事务，并释放数据库连接和 PreparedStatement 资源。

10. 等待下一次调度

1
long cost = System.currentTimeMillis()-start;
2

3
if (cost < 1000) {
4
    try {
5
        TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
6
    } catch (InterruptedException e) {
7
        if (!scheduleThreadToStop) {
8
            logger.error(e.getMessage(), e);
9
        }
10
    }
11
}

如果本次调度耗时小于 1 秒，则线程会休眠一段时间，确保下一次调度在整秒时刻执行。

ringThread#

该线程负责时间轮的推进（单位为秒），并执行对应的任务

1. 线程初始化

1
ringThread = new Thread(new Runnable() {
2
    @Override
3
    public void run() {
4
        // 线程执行的逻辑
5
    }
6
});

2. 线程启动前的延迟

1
try {
2
    TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
3
} catch (InterruptedException e) {
4
    if (!ringThreadToStop) {
5
        logger.error(e.getMessage(), e);
6
    }
7
}

熟悉的操作，线程启动后，首先会休眠一段时间，确保线程在整秒时刻启动（即 System.currentTimeMillis() % 1000 为 0 的时刻）。这样可以使得后续的时间环任务能够对齐到整秒。

3. 主循环

1
while (!ringThreadToStop) {
2
    // 主循环逻辑
3
}

4. 获取当前秒数

1
int nowSecond = Calendar.getInstance().get(Calendar.SECOND);

获取当前的秒数（0-59），用于确定当前时间环的位置。

5. 从时间环中获取任务

1
List<Integer> ringItemData = new ArrayList<>();
2
for (int i = 0; i < 2; i++) {
3
    List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
4
    if (tmpData != null) {
5
        ringItemData.addAll(tmpData);
6
    }
7
}

从时间环数据结构 ringData 中获取当前秒和前一秒的任务列表。
- ringData 是一个以秒为索引的哈希表，存储了每个秒数对应的任务列表。
- (nowSecond + 60 - i) % 60 用于计算当前秒和前一秒的索引，避免跨过刻度。
- 如果某个秒数对应的任务列表存在，则将其合并到 ringItemData 中。

6. 触发任务

1
if (ringItemData.size() > 0) {
2
    for (int jobId : ringItemData) {
3
        JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
4
    }
5
    ringItemData.clear();
6
}

如果 ringItemData 中有任务，则遍历任务列表并触发每个任务。
- JobTriggerPoolHelper.trigger 是触发任务的核心方法，会根据任务 ID 执行相应的任务。
触发完成后，清空 ringItemData，以便下一次使用。

7. 异常处理

1
catch (Exception e) {
2
    if (!ringThreadToStop) {
3
        logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
4
    }
5
}

捕获并记录线程运行过程中发生的异常。如果线程未被要求停止（ringThreadToStop 为 false），则记录错误日志

8. 等待下一次触发

1
try {
2
    TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
3
} catch (InterruptedException e) {
4
    if (!ringThreadToStop) {
5
        logger.error(e.getMessage(), e);
6
    }
7
}

线程休眠一段时间，确保下一次触发在整秒时刻执行。休眠时间通过 1000 - System.currentTimeMillis() % 1000 计算，保证对齐到整秒。

com.xxl.job.admin.core.thread.JobScheduleHelper#toStop

停止时，会先停止预读任务线程，如果有待执行的任务，还会 sleep 8秒等待相应的任务被调度完成

1
public void toStop(){
2

3
    // 1、stop schedule
4
    scheduleThreadToStop = true;
5
    try {
6
        TimeUnit.SECONDS.sleep(1);  // wait
7
    } catch (InterruptedException e) {
8
        logger.error(e.getMessage(), e);
9
    }
10
    if (scheduleThread.getState() != Thread.State.TERMINATED){
11
        // interrupt and wait
12
        scheduleThread.interrupt();
13
        try {
14
            scheduleThread.join();
15
        } catch (InterruptedException e) {
16
            logger.error(e.getMessage(), e);
17
        }
18
    }
19

20
    // if has ring data
21
    boolean hasRingData = false;
22
    if (!ringData.isEmpty()) {
23
        for (int second : ringData.keySet()) {
24
            List<Integer> tmpData = ringData.get(second);
25
            if (tmpData!=null && tmpData.size()>0) {
26
                hasRingData = true;
27
                break;
28
            }
29
        }
30
    }
31
    if (hasRingData) {
32
        try {
33
            TimeUnit.SECONDS.sleep(8);
34
        } catch (InterruptedException e) {
35
            logger.error(e.getMessage(), e);
36
        }
37
    }
38

39
    // stop ring (wait job-in-memory stop)
40
    ringThreadToStop = true;
41
    try {
42
        TimeUnit.SECONDS.sleep(1);
43
    } catch (InterruptedException e) {
44
        logger.error(e.getMessage(), e);
45
    }
46
    if (ringThread.getState() != Thread.State.TERMINATED){
47
        // interrupt and wait
48
        ringThread.interrupt();
49
        try {
50
            ringThread.join();
51
        } catch (InterruptedException e) {
52
            logger.error(e.getMessage(), e);
53
        }
54
    }
55

56
    logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
57
}

核心执行逻辑(版本2.2.0)#

scheduleThread#

ringThread#

文章分享

评论区