轮询设计

最简单的方案

还记得我一开始是怎么实现订单超时取消的吗？

然后用 Linux 的 cron 定时任务，每分钟执行一次：

验证要点

• 命令只用于验证系统状态，读者不需要记具体参数。

这个方案简单粗暴，但真的能用吗？

数据库轮询的工作原理

核心思想

轮询方案的核心：

1. 定期（如每分钟）查询数据库
2. 找出应该执行的任务（执行时间 <= 当前时间）
3. 执行任务
4. 标记任务为已完成

基本实现

轮询器本身就是一个定时循环：每隔一段时间查一次数据库，把到期的任务捞出来执行。

public class DatabasePoller {

    private final int batchSize = 200;

    /** 一轮轮询：捞出到期任务 → 执行 → 标记完成 */
    public void pollOnce() {
        List<DelayTask> tasks = taskDao.findDueTasks(batchSize);
        if (tasks.isEmpty()) {
            return;                                  // 空轮询，注意这里的成本
        }
        for (DelayTask task : tasks) {
            try {
                handlerRegistry.get(task.taskType()).handle(task);
                taskDao.markDone(task.id());
            } catch (Exception e) {
                taskDao.markFailed(task.id(), e.getMessage());
            }
        }
    }
}

配套的 SQL 依赖 execute_at 上的索引：

SELECT id, task_type, task_data, execute_at
FROM delay_tasks
WHERE status = 'PENDING'
  AND execute_at <= NOW()
ORDER BY execute_at
LIMIT 200;

调度既可以用 cron，也可以用代码里的定时器：

ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(this::pollOnce, 1, 1, TimeUnit.SECONDS);

这段代码能跑，但它是”教科书版本”，有三个地方在生产上必须改：

1. 捞取和标记之间没有原子性。 两个轮询线程会捞到同一批任务，导致重复执行。生产版本要用 FOR UPDATE SKIP LOCKED 抢任务，或者用《锁机制》里讲的分布式锁。
2. 每条任务单独提交。 数据库往返次数等于任务数，批量收益为零。生产版本要用《批量处理》里的批量更新和批量占用。
3. 固定间隔轮询。 任务稀疏的时候，绝大多数轮询都是空查询——execute_at 上虽然有索引，但索引只能让”找到”变快，不能让”没东西可查”这件事消失。解决办法见本章最后「想一想」里的方案 1 和方案 2。

数据库表设计

基本表结构

数据设计要点

• 核心是在 delay_tasks 里保存业务事实，而不是把规则散落在应用逻辑里。
• 索引服务于高频查询，重点是缩小扫描范围，而不是堆更多字段。
• 关键字段包括 id、task_id、task_type、task_data、execute_at、retry_count、max_retries、error_message，它们决定后续查询和管理能力。

字段说明

关键索引

数据设计要点

• 索引服务于高频查询，重点是缩小扫描范围，而不是堆更多字段。

为什么这个索引很重要？

数据设计要点

• 索引服务于高频查询，重点是缩小扫描范围，而不是堆更多字段。
• 查询目标是快速定位状态、任务或资源，避免在关键路径上做大范围扫描。

轮询间隔的选择

间隔对精度的影响

自适应轮询

根据任务量动态调整轮询间隔：

固定间隔的问题在于它只能满足一个目标：间隔调小，精度高但数据库压力和 CPU 消耗高；间隔调大，省资源但任务可能晚几十秒才执行。而任务量本身是波动的——白天零星几单，深夜可能一个都没有，固定 1 秒的轮询在深夜纯属浪费。

自适应轮询的核心思路是：根据上一次拉到多少任务，动态决定下一次等多久。

MIN_INTERVAL = 0.5   # 秒，最多 1 秒查一次
MAX_INTERVAL = 30.0  # 秒，最多 30 秒查一次

async def poll_loop():
    interval = MIN_INTERVAL
    while True:
        tasks = await fetch_due_tasks(limit=1000)

        if tasks:
            # 捞到任务说明积压多，立刻用最短间隔再查一次
            interval = MIN_INTERVAL
            for task in tasks:
                await handle(task)
        else:
            # 没捞到说明现在是空闲，逐步拉长间隔
            interval = min(interval * 2, MAX_INTERVAL)

        await asyncio.sleep(interval)

注意这里用了指数退避而不是线性加长：空闲时间越长，间隔增长越快，这样深夜能真正降到 30 秒一次；而一旦来了任务立刻回到 0.5 秒，不会因为退避而错过突发流量。

配套的还有两点：

1. 按状态分开轮询。 待执行任务和已超时的异常任务是两种完全不同的数据，用不同的索引和不同的间隔轮询，异常任务可以被更快地发现。
2. 多实例轮询要错峰。 多个消费者实例同时查同一张表会产生惊群效应——它们会同时查到同一批任务然后互相争抢。用实例 ID 取模错开起始时间，或者用分布式锁保证同一时刻只有一个实例在查，都可以解决。

这套策略的代价是多了一层状态机：间隔要能被观测和调整，空闲时长和任务延迟的关系要有监控。收益是数据库在低峰期几乎不产生压力，而高峰期又能保持秒级精度。

优缺点分析

优点

缺点

想一想

思考 1

如何减少数据库轮询的空查询（没有任务时的查询）？

空查询的代价：

- 数据库连接占用
- 网络往返延迟
- 数据库 CPU 和 I/O 开销
- 日志记录

假设每天 100 万任务，每分钟轮询一次：
- 总轮询次数：1440 次/天
- 如果大部分时间没有任务，90% 是空查询
- 浪费：1296 次无效查询

SELECT MIN(execute_at) FROM delay_tasks WHERE status = 'PENDING';

long nextPoll = Math.max(
        earliestExecuteAt.getTime(),
        System.currentTimeMillis() + 200);   // 兜底最短间隔

// 写入时同时更新缓存
public void submit(DelayTask task) {
    taskDao.insert(task);
    cache.put(task.id(), task.summary());   // 缓存是"待执行任务"的索引
}

// 轮询时
List<TaskSummary> due = cache.dueTasks(System.currentTimeMillis());
if (!due.isEmpty()) {
    List<DelayTask> full = taskDao.findByIds(
            due.stream().map(TaskSummary::id).toList());
    ...
}