《互联网产品的客户端架构·端形态与体验基础》7.3 冲突解决与同步队列

从「回声」在地铁里发出的一条消息出发,把离线写入拆成落盘队列、严格重放、幂等键与冲突合并四步,给出可运行的 TypeScript 队列与九个真实测试,逐条说明 last-write-wins 的边界、字段级合并的判据,并复盘一次超时重试造成重复下单的事故。

离线时点下的“发送”

用户在会话页输入一条消息,点发送,界面立刻显示“发送中”,但网络正好断了。此刻应用面对的不是“请求失败”,而是“请求还没发生”:消息必须留在本地,等网络回来再补发。类似场景还有草稿提交、收藏、地址修改、订单备注,凡是用户在离线状态下表达的写意图,都不能因为一次网络失败而消失。

写路径比读路径难,因为它要处理三个读路径没有的问题:顺序(先创建的实体不能后于它的更新到达)、重复(超时重试可能让同一次写入执行两次)、冲突(离线期间服务端可能已被同一用户或其他人修改)。这一节把三者收进一个队列模型,并给出可运行的实现与测试。

文中出现的时长、版本号与重试次数都是案例假设值,用来演示判断过程,不是任何真实服务的协议。

队列要守住的三个不变量

第一,先落盘再确认。用户点发送后,队列条目必须在 UI 显示“已排队”之前写入存储。进程随后被杀,重启后仍能读回这条写入。反过来,“先发请求,成功后再补日志”在崩溃窗口里会丢数据,而且无法判断请求是否已经到达服务端。

第二,重放顺序稳定。队列按 seq 严格先进先出,跨实体也不重排。这样“创建笔记”一定先于“更新同一笔记”到达服务端。代价是队头阻塞:一个持续失败的条目会挡住后面的条目。案例选择接受这个代价,因为端上队列通常很短,顺序错误造成的数据问题比多等一会儿更严重。

第三,幂等键在入队时生成。键由客户端标识、实体、实体标识与队列序号组成,一旦写入就不再变化。重试、重启、换网络都复用同一个键,服务端据此去重。每次重试重新生成键,等于告诉服务端“这是一次新的写入”,重复就不可避免。

重放顺序与幂等键

重放是一个循环:取队头、增加尝试次数并落盘、发送、根据结果决定删除或保留。网络错误中止本轮,保留剩余条目;服务端返回 duplicate 与 applied 一样处理,因为它说明上一次请求其实已经成功,只是响应丢了。

严格 FIFO 的边界要说清楚:如果队头条目因为服务端 5xx 持续失败,后面的条目会一直等待。生产实现通常加两个约束:单条最大尝试次数,以及“可并行的实体分组”。前者防止死循环,后者允许不同实体的写入并行,但同一实体的写入仍然串行。案例刻意只实现 FIFO,是为了让顺序与幂等这两个不变量先被测试固定下来。

幂等键不是“请求 ID”。请求 ID 通常每次发送都变,用于链路追踪;幂等键标识的是一次业务意图,必须在整个重试生命周期内保持不变。两者可以同时存在,但不能混用。

冲突策略与 LWW 的边界

服务端返回冲突时,客户端必须选一个策略。案例给出三种,按资源的协作强度选择:

策略行为适用
server-wins放弃本地修改,记录冲突权限、配置、他人可编辑的共享资源
client-wins-if-newer本地写入严格晚于服务器时带新版本重放单用户私有数据,允许最后写入者覆盖
merge-disjoint字段无重叠时合并,重叠时回落 LWW多字段表单,双方通常改不同字段

last-write-wins 最容易被误用。它的“最后”必须有一个可信时间源。案例用的是 mutation.queuedAt(本地入队时间)与 server.updatedAt(服务端更新时间)比较,这两个时钟来自不同设备,存在偏移。因此 LWW 只适合“同一用户的私有数据”,并且比较必须是严格大于:本地时间等于服务器时间时判服务器胜。相等意味着无法证明本地更新,宁可不覆盖。

merge-disjoint 的判据是字段级的三方比较:本地 base 里的旧值、本地 payload 里的新值、服务端当前值。如果服务端当前值仍等于本地看到的旧值,说明这个字段没被服务端改过,可以安全写入;如果不等,说明双方都改了同一字段,合并失败,回落到 LWW。合并结果以服务端数据为底稿,只覆盖安全字段,避免把服务端新增的字段删掉。

队列实现

下面的实现把“先落盘、严格 FIFO、稳定幂等键、三种冲突策略”收进一个类。存储层只要求字符串读写,时间从外部注入,测试不需要真实网络或定时器。

// file: src/data/sync-queue.ts
// tier: L1
// verify: node --test test/data/sync-queue.test.ts

export type MutationKind = "create" | "update" | "delete";
export type ConflictPolicy = "server-wins" | "client-wins-if-newer" | "merge-disjoint";
export type ConflictAction = "retry" | "merged" | "drop";

export interface MutationInput {
  entity: string;
  entityId: string;
  kind: MutationKind;
  payload: Record<string, unknown>;
  base: Record<string, unknown>;
  baseVersion: number;
}

export interface Mutation extends MutationInput {
  seq: number; idempotencyKey: string; queuedAt: number; attempts: number; force: boolean;
}

export interface ServerRecord {
  id: string; version: number; updatedAt: number; data: Record<string, unknown>;
}

export type SendResult =
  | { status: "applied"; record: ServerRecord }
  | { status: "duplicate"; record: ServerRecord }
  | { status: "conflict"; record: ServerRecord };

export interface Transport {
  send(mutation: Mutation): Promise<SendResult>;
}

/** 只要求字符串读写,真实实现可以包 localStorage 或 SQLite。 */
export interface QueueStore {
  load(): string | undefined;
  save(raw: string): void;
}

export interface ConflictEvent {
  idempotencyKey: string; strategy: ConflictPolicy; action: ConflictAction;
  serverVersion: number; reason: string;
}

export interface ReplayReport {
  applied: string[]; conflicts: ConflictEvent[];
  remaining: number; stoppedBy: "network" | null;
}

export interface QueueOptions {
  clientId: string; now: () => number; store: QueueStore;
}

interface Resolution {
  action: ConflictAction; payload?: Record<string, unknown>;
  event: ConflictEvent;
}

/**
 * 离线写入队列:先落盘再发送,严格按入队顺序重放。
 * 幂等键在入队时生成并持久化,重试与重启都不会变化。
 */
export class SyncQueue {
  private readonly clientId: string;
  private readonly now: () => number;
  private readonly store: QueueStore;
  private items: Mutation[] = [];
  private seq = 0;

  constructor(options: QueueOptions) {
    this.clientId = options.clientId;
    this.now = options.now;
    this.store = options.store;
    const raw = this.store.load();
    if (raw === undefined) return;
    try {
      const saved = JSON.parse(raw) as { seq: number; items: Mutation[] };
      this.seq = saved.seq;
      this.items = saved.items;
    } catch {
      this.items = [];
      this.seq = 0;
    }
  }

  get size(): number {
    return this.items.length;
  }

  peek(): Mutation | undefined {
    return this.items[0];
  }

  enqueue(input: MutationInput): Mutation {
    const seq = this.seq + 1;
    this.seq = seq;
    const mutation: Mutation = {
      ...input,
      payload: { ...input.payload },
      base: { ...input.base },
      seq,
      idempotencyKey: [
        this.clientId,
        input.entity,
        input.entityId,
        String(seq),
      ].map(encodeURIComponent).join(":"),
      queuedAt: this.now(),
      attempts: 0,
      force: false,
    };
    this.items.push(mutation);
    this.persist();
    return mutation;
  }

  /**
   * 顺序重放。网络错误中止本轮并保留剩余条目;冲突按策略处理。
   * 每次发送前先持久化 attempts,进程在发送途中被杀也能看出尝试过。
   */
  async replay(transport: Transport, policy: ConflictPolicy): Promise<ReplayReport> {
    const applied: string[] = [];
    const conflicts: ConflictEvent[] = [];
    let stoppedBy: "network" | null = null;

    while (this.items.length > 0) {
      const head = this.items[0];
      head.attempts += 1;
      this.persist();

      let result: SendResult;
      try {
        result = await transport.send({ ...head, payload: { ...head.payload }, base: { ...head.base } });
      } catch {
        stoppedBy = "network";
        break;
      }

      if (result.status === "applied" || result.status === "duplicate") {
        applied.push(head.idempotencyKey);
        this.items.shift();
        this.persist();
        continue;
      }

      const resolution = this.resolve(head, result.record, policy);
      conflicts.push(resolution.event);
      if (resolution.action === "drop") {
        this.items.shift();
        this.persist();
        continue;
      }

      const rebased: Mutation = {
        ...head,
        payload: resolution.action === "merged" ? (resolution.payload ?? head.payload) : head.payload,
        base: { ...result.record.data },
        baseVersion: result.record.version,
        force: true,
      };
      this.items[0] = rebased;
      this.persist();

      let retried: SendResult;
      try {
        retried = await transport.send({ ...rebased, payload: { ...rebased.payload }, base: { ...rebased.base } });
      } catch {
        stoppedBy = "network";
        break;
      }

      if (retried.status === "applied" || retried.status === "duplicate") {
        applied.push(rebased.idempotencyKey);
        this.items.shift();
        this.persist();
        continue;
      }

      conflicts.push({
        idempotencyKey: rebased.idempotencyKey,
        strategy: policy,
        action: "drop",
        serverVersion: retried.record.version,
        reason: "重试后仍冲突,停止自动合并",
      });
      this.items.shift();
      this.persist();
    }

    return { applied, conflicts, remaining: this.items.length, stoppedBy };
  }

  private resolve(
    mutation: Mutation,
    server: ServerRecord,
    policy: ConflictPolicy,
  ): Resolution {
    const event = (action: ConflictAction, reason: string): ConflictEvent => ({
      idempotencyKey: mutation.idempotencyKey,
      strategy: policy,
      action,
      serverVersion: server.version,
      reason,
    });

    if (policy === "server-wins") {
      return { action: "drop", event: event("drop", "策略 server-wins,放弃本地变更") };
    }

    if (policy === "client-wins-if-newer") {
      if (mutation.queuedAt > server.updatedAt) {
        return { action: "retry", event: event("retry", "本地写入严格晚于服务器,带新版本重放") };
      }
      return {
        action: "drop",
        event: event("drop", `LWW 边界:本地 ${mutation.queuedAt} 不晚于服务器 ${server.updatedAt}`),
      };
    }

    const merged: Record<string, unknown> = { ...server.data };
    const overlapping: string[] = [];
    for (const [field, value] of Object.entries(mutation.payload)) {
      if (mutation.base[field] === server.data[field]) merged[field] = value;
      else overlapping.push(field);
    }
    if (overlapping.length === 0) {
      return { action: "merged", payload: merged, event: event("merged", "字段无重叠,按字段合并") };
    }
    if (mutation.queuedAt > server.updatedAt) {
      return {
        action: "retry",
        event: event("retry", `字段重叠 ${overlapping.join(",")},回落 LWW 且本地较新`),
      };
    }
    return {
      action: "drop",
      event: event("drop", `字段重叠 ${overlapping.join(",")},回落 LWW 且服务器较新`),
    };
  }

  private persist(): void {
    this.store.save(JSON.stringify({ seq: this.seq, items: this.items }));
  }
}

几个实现细节值得注意。

冲突重试只做一次。 第一次冲突按策略重放;如果重放后仍冲突,直接丢弃并记录原因,不再无限重试。冲突通常意味着有人持续修改同一资源,客户端继续自动合并只会制造更多冲突。丢弃不是“数据没救”,事件已经写进报告,产品可以选择提示用户重新编辑。

重放前先写 attempts。 发送途中进程被杀,重启后能看到这条已经尝试过一次。它不改变幂等键,但为“最大尝试次数”提供了持久化依据。

合并后的 base 更新为服务端当前数据。 这样下一次冲突比较的是最新基线,而不是入队时的旧快照。force 标记告诉传输层“客户端已经做过冲突处理”,服务端可以据此决定是否跳过版本校验,但最终仍由服务端裁决。

传输层返回 duplicate 时不抛错。 这是幂等设计的核心:重复不是异常,而是“上一次其实成功了”。把它当失败重试,才会产生真正的重复。

测试与真实输出

测试覆盖 FIFO、重启持久化、网络中断续传、重复响应、LWW 等值边界、本地更新重放、字段合并、重叠回退与 server-wins 九个场景。传输层是脚本化的,按顺序返回预设结果,不依赖真实服务:

node --test test/data/sync-queue.test.ts

本机一次运行的真实输出如下,耗时随机器变化:

✔ 按入队顺序重放,全部成功后队列清空 (1.523292ms)
✔ 重启后队列与幂等键保持不变 (0.348125ms)
✔ 网络错误中止本轮,剩余条目与幂等键保留 (0.2315ms)
✔ 服务端返回 duplicate 视为成功,避免重复写入 (0.119833ms)
✔ LWW 边界:本地时间等于服务器时间时服务器胜 (0.254125ms)
✔ 本地严格更新时带服务器版本重放 (0.164416ms)
✔ 字段无重叠时合并双方修改 (0.268083ms)
✔ 字段重叠且服务器更新时放弃本地修改 (0.130125ms)
✔ server-wins 策略不重放,直接丢弃本地变更 (0.164041ms)
ℹ tests 9
ℹ suites 0
ℹ pass 9
ℹ fail 0
ℹ cancelled 0
ℹ skipped 0
ℹ todo 0
ℹ duration_ms 357.542167

一次重复下单的事故

某次版本里,下单请求超时后自动重试了一次。第一次请求其实已经到达服务端并创建订单,只是响应在弱网里丢了;客户端重试时没有带幂等键,服务端把它当成第二笔订单。用户回到订单列表,看到两笔相同的待支付订单。

复盘发现三个问题叠加:

  • 客户端把“超时”当成“失败”,没有区分“不确定”。
  • 重试没有幂等键,服务端无法识别这是同一次业务意图。
  • 服务端创建订单的接口不是幂等的,重试必然产生新记录。

修复分三层:客户端在入队时生成稳定幂等键并持久化;服务端在去重窗口内按幂等键返回第一次的结果;产品层在订单列表里合并展示疑似重复的待支付订单,给用户一个取消入口。三层缺一不可,只改客户端不能解决服务端已经重复创建的事实,只改服务端也无法覆盖旧版本客户端。

这个事故说明,离线队列不是“把请求存起来慢慢发”,而是“把业务意图变成可重复提交的幂等操作”。没有幂等键的队列,在网络不确定的环境里只是一个延迟爆炸的重复写入器。

与服务端事务的对照

服务端有事务,客户端没有。两者的对应关系可以列成一张表:

维度服务端事务客户端同步队列
原子性多步操作要么全成要么全败逐条提交,失败保留剩余
隔离性锁与隔离级别无,冲突靠版本与策略
持久性数据库日志本地存储,可能被系统清理
重试通常由上层重试整个事务逐条重放,依赖幂等键
回滚撤销未提交变更只能补偿,不能真正回滚

客户端能承诺的是“最终一致”:每条写入至少提交一次,重复由幂等键消解,冲突由策略裁决。它不能承诺原子性,也不应该假装能回滚。产品设计要接受这个边界:离线期间的多步操作,最终可能只有一部分成功,UI 必须能表达“部分完成”。

小结

离线写入队列守住三个不变量:先落盘再确认、严格顺序重放、幂等键入队即固定。三者共同保证“至少一次提交,重复可消解”。冲突策略是第二层:server-wins 适合共享资源,client-wins-if-newer 适合单用户私有数据,merge-disjoint 适合多字段表单;LWW 必须使用严格大于,并且只在可信时间源范围内使用。

队列不能替代服务端事务。它把“网络不确定”转换成“可重放的业务意图”,剩下的原子性、隔离性与去重要由服务端配合完成。下一章从数据层转向组件体系,讨论设计系统如何把端上的交互契约固定下来。

阅读导航:上一节:7.2 端上缓存与离线优先 · 下一节:8.1 设计系统与组件契约 。

常见错误与边界

  • 每次重试重新生成幂等键。服务端会把它当成新的写入,重复无法避免。
  • 队列只放内存。进程被系统回收后,用户离线期间的写入全部丢失。
  • 先发送成功再补日志。崩溃窗口里既不知道请求是否到达,也无法恢复。
  • 冲突时直接覆盖服务端。可能抹掉其他人的修改,且无法审计。
  • 用客户端时钟做全局 LWW。设备时间可被修改,跨用户比较没有意义。
  • 合并时用本地快照整体替换。服务端新增的字段会被静默删除。
  • 重放非幂等接口且不做服务端去重。超时重试必然产生重复记录。
  • 忽略队头阻塞。一个持续失败的条目会让整个队列停摆。
  • 把队列当事务。离线多步操作可能部分成功,UI 必须能表达。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《互联网产品的 AI 架构·平台化与治理》附录 C 平台化评审清单
  2. 《互联网产品的 AI 架构·平台化与治理》附录 B 能力目录与配额字段速查
  3. 《互联网产品的 AI 架构·平台化与治理》附录 A 治理角色与职责速查