缓存、队列与后台任务

本章解决什么问题#

文档摘要生成需要二十秒，浏览器等待容易超时；十个人同时打开同一份摘要又触发十次计算；服务器重启后，内存中的“稍后执行”任务全部消失。本章目标是分别解决重复读取和耗时执行，理解缓存、队列、工作者与数据库事实来源的边界。前置是并发控制、事务、租户授权与文件状态。

缓存回答“已有结果能否复用”，队列回答“已接受的工作由谁在什么时候执行”。两者都可能使用 Redis，但不能因此混为一谈。缓存丢失通常可以从事实来源重建；已经确认接受的作业丢失，则可能违背产品承诺。不要把一个普通 Map 或 setTimeout 当成可靠后台队列。

缓存先定义可接受的陈旧程度#

最简单的读取流程是先查缓存，未命中再读数据库并回填。缓存键必须包含决定结果的全部维度，例如租户、文档版本、语言以及必要的权限视图。只用文档 id 缓存管理员可见内容，可能把私有结果复用给普通成员。使用版本键往往比试图找出所有旧结果逐个删除更清楚，但版本变更与权限撤销仍需要明确策略。

TTL 限制结果最多保存多久，不保证写入后立刻一致。容量上限限制内存或存储增长；请求合并让同一个键的并发未命中共享一次加载，缓解热门键同时回源。失败结果是否缓存、空结果是否短暂缓存、缓存不可用时是否降级，都是可以独立决定的合同。

Redis 的 SET 可用 EX 秒或 PX 毫秒设置过期时间，也可用 NX 表示仅当键不存在时写入；这些选项有明确返回语义。但一个带过期时间的 NX 锁，仍不等于拥有完整的分布式互斥保证，释放时必须核对所有者，过期后旧工作者可能继续运行。Redis SET 是命令的官方合同。

完整示例一：有容量上限的本地缓存与请求合并#

环境：Node 22.22 或 24，无依赖。保存为 cache.mjs，运行 node cache.mjs。本例是单进程缓存，不能保证多个服务实例共享结果；同一键的返回值使用字符串，避免可变对象被多个调用方意外修改。

// cache.mjs
import assert from 'node:assert/strict';
import { setTimeout as delay } from 'node:timers/promises';

const cache = new Map();
const inflight = new Map();
const MAX_ENTRIES = 100;
const MAX_PENDING = 20;

async function getOrLoad(key, loader, ttlMs = 50) {
  if (!Number.isInteger(ttlMs) || ttlMs < 1 || ttlMs > 60000) {
    throw new RangeError('TTL 不合法');
  }
  const hit = cache.get(key);
  if (hit && hit.expiresAt > Date.now()) return hit.value;
  cache.delete(key);
  if (inflight.has(key)) return inflight.get(key);
  if (inflight.size >= MAX_PENDING) throw new Error('回源任务已满');

  const pending = Promise.resolve().then(loader).then((value) => {
    // 简单先进先出淘汰，不声称这是 LRU。
    if (cache.size >= MAX_ENTRIES) cache.delete(cache.keys().next().value);
    cache.set(key, { value, expiresAt: Date.now() + ttlMs });
    return value;
  }).finally(() => inflight.delete(key));
  inflight.set(key, pending);
  return pending;
}

let loads = 0;
const key = 'tenant:1:document:42:version:3:viewer:20';
async function load() {
  loads += 1;
  await delay(10);
  return `摘要版本-${loads}`;
}
const results = await Promise.all([
  getOrLoad(key, load), getOrLoad(key, load), getOrLoad(key, load),
]);
assert.equal(loads, 1);
assert.deepEqual(results, ['摘要版本-1', '摘要版本-1', '摘要版本-1']);
await delay(70);
assert.equal(await getOrLoad(key, load), '摘要版本-2');
assert.equal(loads, 2);
assert.equal(inflight.size, 0);
console.log('三个并发读取只回源一次，TTL 到期后重新加载');

预期输出一条请求合并与过期验证说明。pending 在实际 loader 的微任务启动前登记，后续请求才能找到它；finally 无论成功失败都移除占位，避免一次失败让这个键永远卡住。MAX_PENDING 还限制了不同键同时未命中的数量，缓存容量与回源容量是两个不同资源上限。

后台任务需要明确状态与领取权#

提交接口应先持久化任务，再返回 202 与任务 id；202 表示接受处理，不表示解析已经成功。工作者把 pending 任务领取为 running，成功后写 done，失败后按规则重试或进入 failed。页面可以轮询任务状态，也可用 SSE 接收变化；状态接口仍要做租户与资源授权。

领取权通常有有效期，称为租约。工作者崩溃后，过期任务可再次被领取；因此同一个任务可能执行多次，这要求结果写入幂等。租约令牌用于判断“完成报告是否仍来自当前领取者”，旧工作者醒来后不能覆盖新工作者的结果。长任务还需续租、心跳和取消策略，不能只把租期随手写成一个很大的常量。

FOR UPDATE SKIP LOCKED 可以跳过被其他工作者锁住的候选行，适合队列领取，但会得到不完整视图，不适用于一般报表“跳过忙行就算全量结果”。领取后不要持有事务锁等待模型调用；先提交领取状态，外部计算完成后再开短事务写结果。PostgreSQL SELECT 专门说明了 SKIP LOCKED 的使用边界。

完整示例二：领取、一次重试与幂等完成#

环境：Node 22.22 或 24、PostgreSQL 16+、pg 8。练习目录执行 npm.cmd init -y、npm.cmd install pg@8，设置 DATABASE_URL，保存为 queue.mjs 后执行 node queue.mjs。为便于重复练习，以下使用连接级临时表；临时表本身不具备跨重启持久性。正式服务必须使用迁移创建的持久表、共享连接与监控，不能原样部署本练习来承诺任务不丢失。

// queue.mjs
import pg from 'pg';
import { randomUUID } from 'node:crypto';
import { setTimeout as delay } from 'node:timers/promises';
import assert from 'node:assert/strict';

if (!process.env.DATABASE_URL) throw new Error('请设置 DATABASE_URL');
const db = new pg.Client({ connectionString: process.env.DATABASE_URL });
await db.connect();

async function claim() {
  // 达到重试上限且租约过期的任务转入终止失败状态。
  await db.query(`UPDATE lesson_jobs SET state = 'failed'
    WHERE state = 'running' AND locked_until < clock_timestamp() AND attempts >= 3`);
  const result = await db.query(`WITH candidate AS (
    SELECT id FROM lesson_jobs
    WHERE attempts < 3 AND (
      (state = 'pending' AND run_after <= clock_timestamp()) OR
      (state = 'running' AND locked_until < clock_timestamp())
    )
    ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED
  )
  UPDATE lesson_jobs AS j SET state = 'running', attempts = attempts + 1,
    lease_token = $1, locked_until = clock_timestamp() + interval '30 seconds'
  FROM candidate AS c WHERE j.id = c.id RETURNING j.*`, [randomUUID()]);
  return result.rows[0] ?? null;
}

async function finish(job, output) {
  await db.query('BEGIN');
  try {
    // 同一事务确认领取权，写结果，再标记完成。
    const owned = await db.query(`SELECT id FROM lesson_jobs
      WHERE id = $1 AND lease_token = $2 AND state = 'running'
        AND locked_until > clock_timestamp() FOR UPDATE`, [job.id, job.lease_token]);
    if (owned.rowCount !== 1) throw new Error('领取权已经失效');
    await db.query(`INSERT INTO lesson_results (job_id, tenant_id, output)
      VALUES ($1, $2, $3) ON CONFLICT (job_id) DO NOTHING`,
      [job.id, job.tenant_id, output]);
    await db.query(`UPDATE lesson_jobs SET state = 'done', locked_until = NULL
      WHERE id = $1 AND lease_token = $2`, [job.id, job.lease_token]);
    await db.query('COMMIT');
  } catch (error) {
    await db.query('ROLLBACK');
    throw error;
  }
}

try {
  await db.query(`
    CREATE TEMP TABLE lesson_jobs (
      id integer GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
      tenant_id integer NOT NULL, payload jsonb NOT NULL,
      state text NOT NULL DEFAULT 'pending'
        CHECK (state IN ('pending', 'running', 'done', 'failed')),
      attempts integer NOT NULL DEFAULT 0,
      run_after timestamptz NOT NULL DEFAULT now(),
      locked_until timestamptz, lease_token uuid, last_error text
    );
    CREATE TEMP TABLE lesson_results (
      job_id integer PRIMARY KEY, tenant_id integer NOT NULL, output text NOT NULL
    );
  `);
  for (const payload of [{ text: 'document A', failOnce: false },
    { text: 'document B', failOnce: true }]) {
    await db.query('INSERT INTO lesson_jobs (tenant_id, payload) VALUES ($1, $2::jsonb)',
      [1, JSON.stringify(payload)]);
  }
  // 有界循环便于练习退出；正式 worker 还需停止信号和健康检查。
  for (let tick = 0; tick < 100; tick += 1) {
    const job = await claim();
    if (!job) {
      const unfinished = await db.query(`SELECT count(*)::integer AS count
        FROM lesson_jobs WHERE state IN ('pending', 'running')`);
      if (unfinished.rows[0].count === 0) break;
      await delay(20);
      continue;
    }
    try {
      await delay(10); // 模拟外部工作，不在数据库事务中等待。
      if (job.payload.failOnce && job.attempts === 1) throw new Error('模拟瞬时故障');
      if (typeof job.payload.text !== 'string') throw new Error('任务文本不合法');
      await finish(job, job.payload.text.toUpperCase());
    } catch (error) {
      // 比较领取令牌，旧工作者不能改写后来重新领取的任务。
      await db.query(`UPDATE lesson_jobs
        SET state = CASE WHEN attempts >= 3 THEN 'failed' ELSE 'pending' END,
            run_after = clock_timestamp() + interval '100 milliseconds',
            locked_until = NULL, last_error = $3
        WHERE id = $1 AND lease_token = $2 AND state = 'running'`,
        [job.id, job.lease_token, error.message.slice(0, 200)]);
    }
  }
  const jobs = (await db.query('SELECT state, attempts FROM lesson_jobs ORDER BY id')).rows;
  assert.deepEqual(jobs, [{ state: 'done', attempts: 1 }, { state: 'done', attempts: 2 }]);
  const count = (await db.query('SELECT count(*)::integer AS count FROM lesson_results')).rows[0].count;
  assert.equal(count, 2);
  console.log('两个任务完成，第二个重试一次，结果表只有两条');
} finally {
  await db.end();
}

预期输出两个任务完成的说明。这个练习只验证一个工作者的状态流转和 SQL 结构，未模拟多进程崩溃恢复。虽然领取语句使用了适合多工作者的锁策略，不能据此宣称已经完成分布式压力与故障验收。

关键代码与真实边界#

claim 用一个 SQL 完成选行与改状态，避免两个工作者先读到同一条 pending 再分别领取。finish 在短事务中比较 lease_token、写唯一 job_id 结果和更新 done。它保护的是同一数据库里的结果，不会把外部模型调用变成“恰好执行一次”；供应商请求仍可能重复，需要业务幂等键、可恢复结果或可接受的重复成本。

例子使用固定一百毫秒重试以便观察。实际系统应区分可重试错误与永久错误，使用指数退避和抖动，设置总截止时间，并保留失败原因与人工重试入口。队列深度、最老任务等待时间、执行耗时、重试率和失败率比一个简单的“服务在线”灯更能反映用户是否在等待。

从用户等待看缓存与任务的不同职责#

“团队文档与任务助手”打开一篇已处理文档时，可以复用已有摘要；上传新版本时，却必须接受并跟踪一项新的解析工作。前者允许在缓存丢失后重新读取事实来源，后者在向用户确认接受以后不能悄悄遗忘。把两者都放进 Redis，不会自动让它们拥有相同可靠性，存储介质的名字不能替代业务保证。

缓存命中应返回哪个版本、允许陈旧多久、权限变化后如何失效，都是读取合同。任务提交则要说明是否已经持久化、是否可能重复执行、失败如何重试、结果保留多久。前端的已接受提示与任务完成提示应分别出现；如果用户关闭标签页，已提交任务通常继续存在，而临时预览可以按合同取消。

最初可以把队列实现放在 PostgreSQL 中，复用现有事务与监控；当吞吐、延迟、调度能力或组织边界需要时，再选择专门消息系统。独立队列引入新的部署、权限、积压和投递语义，不能只因为某个库容易安装就认为架构变简单。反过来，数据库队列表也会增加轮询与写入压力，需要实际负载评估。

缓存键是结果身份的一部分#

摘要不仅取决于 document_id，还可能取决于文档版本、提示模板版本、模型配置、目标语言与用户可见范围。如果其中某个维度改变却仍复用旧键，缓存命中反而成为错误来源。可以把这些稳定输入编码成一个可解释的键或摘要，但不要直接把完整私有提示词放进容易被运维工具查看的键名。

权限不能只在回源时检查。缓存中的内容可能由管理员首次加载，普通成员随后命中时如果跳过授权，就会绕过整个数据库过滤。可以在读取前执行授权，再使用与权限视图对应的缓存；也可以缓存不含用户私有差异的底层公开片段，让服务层组合。选择取决于数据模型，不能用一个全局 document:id 键覆盖所有情况。

TTL 限制时间，容量策略限制空间，主动失效处理写入变化，请求合并限制同键回源，四者解决不同问题。只有 TTL 的 Map 仍可能在到期记录从不再访问时增长；只有容量上限也可能持续返回很旧的结果。教学缓存使用简单先进先出淘汰，是有意明确的策略，不应被误称为已经实现完整 LRU 或分布式一致性。

最小实验：删除缓存仍可能被旧请求回填#

一次请求读取旧文档后开始耗时计算，另一次请求更新文档并删除缓存，旧计算随后完成又把旧摘要写回同一个键。这就是为什么“写数据库后删缓存”仍要考虑并发。版本化键让旧计算只写旧版本，不覆盖新版本的读取身份。

保存为 cache-version.mjs，Node 22.22 或 24 执行 node cache-version.mjs，无依赖。程序用可控 Promise 屏障制造交错，不依赖模糊的定时碰运气；它只展示版本隔离，生产仍需容量与过期管理。

// cache-version.mjs
import assert from 'node:assert/strict';
let document={tenantId:1,id:42,version:1,text:'旧正文'};
const cache=new Map();let releaseOld;
const oldGate=new Promise(resolve=>{releaseOld=resolve;});
async function readSummary(){
  const snapshot={...document};
  const key=`tenant:${snapshot.tenantId}:doc:${snapshot.id}:version:${snapshot.version}`;
  if(cache.has(key))return cache.get(key);
  if(snapshot.version===1)await oldGate;
  const value=`摘要:${snapshot.text}`;
  cache.set(key,value);return value;
}
const oldRequest=readSummary();
document={...document,version:2,text:'新正文'};
assert.equal(await readSummary(),'摘要:新正文');
releaseOld();assert.equal(await oldRequest,'摘要:旧正文');
assert.equal(await readSummary(),'摘要:新正文');
assert.equal(cache.size,2);
console.log('旧请求只回填版本 1，新请求持续读取版本 2');

预期输出一条说明。旧请求返回它开始时的版本并不自动构成错误，前端需要知道结果版本并决定是否仍展示；真正危险的是把旧结果标成最新。版本键把这种关系显式化，也带来旧版本清理成本。删除文档时应使权限与当前版本读取失效，不能仅等待所有历史缓存自然过期。

API 细读：加载函数、过期与失败#

本章 getOrLoad 的输入是 key、无参数 loader 与可选 ttlMs，默认五十毫秒只是为了快速实验；生产应按数据可陈旧程度设定，不照抄这个数值。返回 Promise，命中时得到已有值，未命中等待 loader；loader 拒绝时不会缓存成功值，并在 finally 中移除 in-flight 占位。不同键同时未命中还受 MAX_PENDING 限制，达到上限返回忙碌而不是无限排队。

Redis SET 的 EX 以秒计，PX 以毫秒计，单位混淆可能让缓存寿命差一千倍。NX 表示仅在键不存在时写入，条件不满足时的返回值必须检查；收到响应不能一律认为已经获得锁。锁值应标识拥有者，释放时需要原子比较并删除，不能先 GET 比较再独立 DEL，否则中间可能已经换了新的拥有者。

队列 claim 返回一个已获得租约的任务或 null，null 表示当前没有可领取任务，不等于所有任务都已经完成；可能有任务还在运行、等待退避或被其他工作者领取。finish 应验证任务编号、租约令牌与当前状态，返回是否确实提交完成。教学脚本的三次尝试和三十秒租期是显式策略，并不是 PostgreSQL 自带的默认队列语义。

至少一次执行如何影响业务实现#

工作者可能在计算成功后、保存结果之前崩溃，也可能在保存结果以后、确认消息之前失去连接。系统若要恢复未确认工作，就可能再次执行同一任务。因此任务处理不能假设只调用一次。生成文档摘要可以按版本覆盖同一个结果槽，扣费或发送通知则需要业务幂等记录，不能因为队列有 jobId 就自动认为外部系统不会重复接受操作。

任务编号与尝试编号应分开。任务编号标识用户提交的一次意图，重试沿用它；每次领取产生新的租约或尝试标识，便于区分哪个工作者有权提交。失败重试时重新生成整个业务 id，会破坏下游去重。手动重跑如果代表新的业务意图，应显式创建新任务并说明计费与结果关系，而不是偷偷绕过原任务状态。

租约过期也不等于旧工作者真正停止。它可能仍在等待模型响应，随后拿到结果；因此新工作者重新领取以后，旧令牌的完成报告必须被拒绝。只有任务表里的租约检查还不够保护外部副作用，外部接口若支持幂等键，应使用稳定业务键，必要时先查询已经产生的结果。

原练习的完整参考：失败上限与旧领取者拒绝#

保存为 queue-state-lab.mjs，Node 22.22 或 24 执行 node queue-state-lab.mjs，无依赖。本实验用内存状态机和显式时钟完成原练习的失败次数、其他任务继续执行与旧领取令牌拒绝检查。它是可重复的算法实验，不具备数据库持久性，也不声称完成了多进程故障验收；对应生产 SQL 仍使用本章前面的队列示例。

完整代码已收录在本章末尾的练习参考答案中；可先阅读说明，再展开复制运行。

显式时钟让测试直接跨过租约期限，不必真的等待三十秒。真实分布式实现应考虑时钟来源，数据库队列通常可以使用数据库时间减少工作者时钟差异。状态机实验验证的是转移规则，数据库测试还要验证锁、事务和多连接竞争；两个层次相互补充，不能只留其中一个。

重试、死信与人工恢复#

错误分类应决定是否重试。暂时网络失败可退避，文档格式永远不支持则应尽快失败，额度不足可能需要等待用户补充资源而不是高速重复调用。退避通常逐步增加，并加入抖动，防止许多任务同时恢复时形成峰值。总尝试次数和总存活时间都要有上限，避免任务长期占住队列又不给用户明确结果。

失败任务应保存可公开的原因与内部诊断标识，人工恢复入口要显示原始输入版本、已产生的副作用和之前尝试。不能为了“让队列变绿”直接删除失败记录，也不能无条件重放所有历史失败，可能产生重复费用或处理已删除文档。恢复本身是一项受授权与审计的业务操作。

如果任务分成上传验证、文本提取、向量生成和发布索引，可以为阶段保存检查点。重试只重做未完成且可幂等的阶段，减少重复成本；但检查点必须与实际产物一致，不能在产物写完前就标记阶段完成。复杂度随着阶段增加，应先定义恢复要求，再决定是否需要完整工作流引擎。

任务取消、删除与版本变更#

用户取消任务是一种意图，需要工作者在合适边界观察。正在进行的外部请求可能不能撤销，因此取消状态与已产生费用应分别记录。一个任务已被取消，晚到结果不应自动发布为最新文档；finish 时还要检查取消和版本状态。只在前端停止轮询不会取消后台工作。

文档被删除后，排队中的任务可能仍存在。工作者执行前应根据业务策略检查资源当前状态，避免重新生成用户已经删除的内容。若删除流程要求清除派生产物，任务发布步骤也应受版本或墓碑条件限制，防止清理完成后旧任务又把文件写回来。队列与文档生命周期必须共用明确身份与状态。

调试与容量规划#

队列深度只是一个指标，还应观察最老等待时间、领取速率、完成速率、每次执行耗时、重试比例与失败类别。队列很短但每项卡住一小时，用户体验仍然很差；队列很长但持续快速完成，可能只是正常批量导入。将这些指标按租户和任务类型分开，才能看见少数用户被大批量工作挤压的情况。

工作者并发应受 CPU、内存、连接池和供应商配额共同限制。增加进程数量会提高总并发，也会增加数据库连接与模型调用额度，不能只修改队列库中的 concurrency 后就认为容量无限扩大。针对大 PDF 和小文本分别测量，更能反映真实资源差异。

关闭工作者时先停止领取新任务，再等待已领取任务在截止时间内完成；无法完成时按租约或检查点交还。不要立即把 running 全部改回 pending 而旧计算仍在继续，否则容易产生重复副作用。健康检查也应区分进程存活与工作者实际能连接数据库、领取任务和提交结果。

业务事务与队列的接缝#

创建文档记录后直接向外部队列发送消息，会遇到数据库与消息系统不同步的问题。可以在同一数据库事务写入任务或 outbox，提交后转发，并让消费者处理重复。任务状态查询读取事实来源，不能只查询消息是否还在队列里：消息被领取以后可能消失，但任务仍然运行；消息确认以后，用户仍需要历史结果。

缓存也不应成为任务完成的唯一记录。摘要缓存被淘汰以后，系统应能从持久结果恢复，而不是重新计费执行一个已经完成的任务。把事实来源、加速副本和执行通知区分清楚，会让故障恢复、用户查询与成本核对更容易成立。这是团队文档助手能够长期可靠运行的基础。

练习：让重试永远不会无限进行#

把第二项改成每次都失败，要求三次后进入 failed；再加入第三个正常任务，验证坏任务不会阻塞其他任务。提示：状态、attempts 和 run_after 一起决定是否可领取，不要只在 catch 中打印错误后马上无限循环。

// queue-state-lab.mjs
import assert from 'node:assert/strict';
function createQueue(){
  const jobs=[];const results=new Map();let sequence=0;let leases=0;
  function enqueue(payload){const row={id:++sequence,payload,state:'pending',attempts:0,runAfter:0,until:0,lease:0};jobs.push(row);return row.id;}
  function claim(now){
    for(const row of jobs){
      if(row.state==='running'&&row.until<=now&&row.attempts>=3)row.state='failed';
      const available=(row.state==='pending'&&row.runAfter<=now)||(row.state==='running'&&row.until<=now);
      if(!available||row.attempts>=3)continue;
      row.state='running';row.attempts+=1;row.until=now+30;row.lease=++leases;
      return {...row}; // 工作者拿到快照，不能直接改共享记录。
    }
    return null;
  }
  function owned(ticket,now){return jobs.find(row=>row.id===ticket.id&&row.lease===ticket.lease&&row.state==='running'&&row.until>now);}
  function finish(ticket,value,now){const row=owned(ticket,now);if(!row)return false;results.set(row.id,value);row.state='done';return true;}
  function fail(ticket,now){const row=owned(ticket,now);if(!row)return false;row.state=row.attempts>=3?'failed':'pending';row.runAfter=now+5;return true;}
  return {enqueue,claim,finish,fail,jobs,results};
}
const queue=createQueue();queue.enqueue({alwaysFail:true});queue.enqueue({alwaysFail:false});
for(let now=0;now<30;now+=1){
  const ticket=queue.claim(now);if(!ticket)continue;
  if(ticket.payload.alwaysFail)queue.fail(ticket,now);else queue.finish(ticket,'已处理',now);
}
assert.deepEqual(queue.jobs.map(row=>[row.state,row.attempts]),[['failed',3],['done',1]]);
assert.equal(queue.results.size,1);
const recovery=createQueue();recovery.enqueue({});
const old=recovery.claim(0);const replacement=recovery.claim(31);
assert.notEqual(old.lease,replacement.lease);
assert.equal(recovery.finish(old,'旧结果',32),false);
assert.equal(recovery.finish(replacement,'新结果',32),true);
assert.equal(recovery.results.get(old.id),'新结果');
console.log('坏任务三次后失败，好任务正常完成，旧领取者不能覆盖新结果');

可验证的验收标准#

同键并发未命中只回源一次，TTL 到期重新加载，缓存和回源都有上限；任务提交与执行结果可区分；失败有上限；旧领取令牌不能覆盖新领取；结果写入有唯一约束；缓存键和作业记录带租户上下文。正式交付还必须用持久表执行重启、租约过期和多工作者实验。

三个自测问题与答案#

1. setTimeout 能作为可靠延迟任务队列吗？答案：不能，进程退出后内存计时器与待执行函数会丢失。
2. SKIP LOCKED 适合拿完整财务报表吗？答案：不适合，它可能跳过当前锁住的行，适合领取队列等允许不完整候选视图的场景。
3. 数据库结果只写一次是否证明模型只调用一次？答案：不能，外部副作用不属于本地事务，必须单独处理重复调用与幂等。

本章验证记录#

编写时使用 Node 22.22.0 对本章全部 4 个 JavaScript 完整文件执行了语法检查。已在本机实际运行通过：cache.mjs、cache-version.mjs、queue-state-lab.mjs。以下文件未连接 PostgreSQL 实际执行，SQL 行为、查询计划与并发保证仍需在专用练习数据库验证：queue.mjs。内存状态机实验不具备跨进程持久性，尚未实际验证多工作者崩溃恢复。

本章官方参考#

• Redis SET：条件写入与过期选项。
• PostgreSQL SELECT：FOR UPDATE 与 SKIP LOCKED。
• PostgreSQL Explicit Locking：行锁与并发控制。
• node-postgres Transactions：短事务的连接边界。