实现幂等性处理，忽略已处理的任务

识别已处理的任务

忽略已处理的任务，听起来只需要一个简单的检查：任务处理过了吗？处理过就跳过，没处理过就执行。

但这个"简单检查"藏着复杂性。

问题不在于检查本身，而在于如何确定一个任务真的处理过了。

你可以用一个布尔字段 processed 来标记。但这个字段什么时候设为 true？执行操作之前还是之后？如果在执行前设为 true，操作失败了怎么办？如果在执行后设为 true，并发请求怎么办？

两个请求同时检查，都发现任务未处理，于是都去执行。检查没有意义。

这不是实现的问题，而是时序的问题。检查和执行之间存在时间窗口，这个窗口里可能发生任何事情。

原子性是关键

要真正忽略已处理的任务，检查和标记必须是原子的。

要么你成功标记了这个任务为"处理中"，然后去执行；要么标记失败，说明别人已经在处理了，你直接返回。

不能先检查再标记。检查和标记必须在同一个不可分割的操作里完成。

数据库的原子操作可以做到这一点：

-- 尝试将未处理的任务标记为处理中
UPDATE tasks
SET status = 'processing'
WHERE task_id = ?
  AND status = 'pending'

这个 UPDATE 语句是原子的。如果 status 已经不是 pending 了，UPDATE 不会成功。你可以检查 affected_rows，如果是 0，说明任务已经被处理或正在被处理。如果是 1，说明你成功获得了处理权。

这是一种乐观锁的思路。不要先问能不能做，直接去做，看做成了没有。

if (task.status === 'pending') {
  task.status = 'processing';
  await processTask(task);
}

const updated = await db.update({
  where: { id: task.id, status: 'pending' },
  data: { status: 'processing' }
});

if (updated.count === 0) {
  // 任务已被处理，直接返回
  return;
}

await processTask(task);

状态标记的选择

用什么来标记任务已处理？

最简单的是布尔值 processed。但布尔值只有两个状态：处理过和没处理过。它无法区分"正在处理"和"处理完成"。

如果一个任务正在被处理，但处理到一半系统崩溃了，这个任务应该怎么办？如果标记是 processed = true，它会被永久忽略。如果标记是 processed = false，它会被重复执行。

你需要更细的粒度。

一个状态字段可以有多个值：pending、processing、completed、failed。

• pending：等待处理
• processing：正在处理
• completed：处理成功
• failed：处理失败

有了这些状态，你可以做更智能的判断。

completed 的任务应该被忽略。processing 的任务要看情况——如果它处于这个状态太久了，可能是之前的处理者崩溃了，应该重新处理。failed 的任务可能需要重试。

状态不只是标记，它携带了信息。它告诉你这个任务经历了什么，现在处于什么情况。

时间戳的作用

状态字段告诉你任务处于什么阶段，但它不告诉你任务在这个阶段待了多久。

这就是时间戳的价值。

updated_at 记录了状态最后一次变更的时间。如果一个任务的状态是 processing，但 updated_at 是 10 分钟前，那很可能之前的处理者已经挂了。

你可以写一个定时任务，把这些"卡住"的任务重置为 pending，让它们重新被处理。

UPDATE tasks
SET status = 'pending'
WHERE status = 'processing'
  AND updated_at < NOW() - INTERVAL 5 MINUTES

这是一种自愈机制。系统不需要人工介入，就能从异常状态中恢复。

但要注意，这个超时时间不能太短。如果一个任务真的需要处理 5 分钟，你把超时设成 3 分钟，它会被反复重置，永远处理不完。

超时时间应该基于正常处理时间的最大值，而不是平均值。

唯一标识与去重

有时状态字段不够，你需要更明确的标识来判断重复。

比如消息队列。同一个消息可能因为网络问题被投递多次。你不能依赖消息队列的去重机制，因为它可能不存在或不可靠。

你需要给每个消息一个唯一的 ID，然后在处理时检查这个 ID 是否已经处理过。

async function processMessage(message) {
  const messageId = message.id;

  // 尝试插入消息 ID
  try {
    await db.insert({
      table: 'processed_messages',
      data: { message_id: messageId, processed_at: new Date() }
    });
  } catch (error) {
    if (error.code === 'UNIQUE_VIOLATION') {
      // 消息已处理，直接返回
      return;
    }
    throw error;
  }

  // 执行实际处理
  await handleMessage(message);
}

这里的关键是先插入 ID，再处理消息。插入失败说明已处理过，直接返回。插入成功才去处理。

这个模式依赖数据库的唯一约束。它把"是否处理过"的判断交给了数据库，而不是应用层的逻辑。

数据库的约束是可靠的，应用层的逻辑是脆弱的。把关键决策放在可靠的层面。

清理的必要性

无论是状态标记还是消息 ID，它们都会积累。

一个高流量的系统，每天可能处理百万级的任务或消息。如果你永久保存所有的处理记录，存储会成为问题，查询也会变慢。

你需要定期清理已完成的记录。

但清理多久之前的记录？这取决于你的重复窗口有多大。

如果消息队列的重试策略是"失败后每隔 1 分钟重试一次，最多重试 10 次"，那么一个消息最多会在 10 分钟内重复。你可以安全地删除 1 小时前的处理记录。

如果系统有人工重试，或者有延迟很久的异步通知，你可能需要保存更久——比如 7 天或 30 天。

清理策略反映了你对系统行为的理解。 你知道重复会在什么时间范围内发生，你就知道记录需要保存多久。

-- 按日期分区
CREATE TABLE processed_tasks (
  task_id VARCHAR(255),
  processed_at TIMESTAMP,
  ...
) PARTITION BY RANGE (processed_at);

-- 定期删除旧分区
DROP TABLE processed_tasks_2024_10;

不只是防止重复

忽略已处理的任务，表面上是为了防止重复执行。

但更深层的意义是：让系统对混乱有韧性。

消息会重复，网络会超时，进程会崩溃。如果系统假设一切正常，它会在异常情况下崩溃。如果系统预期混乱，它会在混乱中保持稳定。

实现幂等性处理，就是在告诉系统：重复是正常的，失败是常态，不要惊慌，知道如何应对。

这不只是一个技术细节，而是一种设计哲学。

你不是在构建一个只能在完美环境下运行的系统，你是在构建一个能在不完美环境下可靠运行的系统。

参考与延伸