【自学AI】12 幂等性：让你的接口不怕用户乱点

4149 字
21 分钟
【自学AI】12 幂等性：让你的接口不怕用户乱点
2026-03-16
AI
SSE
/
langChain.js
/
NestJS
浏览量 加载中...

上一篇，我们搭好了 SSE 流式推送的框架——用户发起请求，后端实时推送进度，前端进度条跟着动。

但上线之前，有一个问题必须先解决：

如果用户不小心点了两次”生成”按钮，怎么办？

开头：一个真实的麻烦#

先说一个我自己遇到过的场景。

测试接口的时候，网络有点慢，等了三秒没反应，就又点了一次。 结果两个请求都跑完了——AI 被调用了两次，数据库里存了两条记录，配额扣了两次。

这不是用户的错。 用户不知道后台在处理，他只是不确定有没有成功，所以再点了一次。

这种情况在真实产品里，比你想象的更常见：

  • 用户等得不耐烦，连续点了好几次”生成”按钮
  • 网络不稳定，客户端自动重试了请求
  • 鼠标左键有问题，单击变成了双击

如果傻傻地执行每一个请求，后果是：

  • ❌ 浪费 AI 调用次数（每次都要扣费）
  • ❌ 浪费用户配额
  • ❌ 浪费服务器资源
  • ❌ 用户体验很差（等了 10 秒生成完，结果又生成了一遍）

解决这个问题的技术，叫做幂等性检查。

这篇文章，就来搞清楚：幂等性是什么、为什么需要它、以及怎样在代码里实现它。

一、什么是幂等性？#

先从数学说起#

幂等性（Idempotency）这个词来自数学。定义是：

对于某个操作 ff，如果满足 f(f(x))=f(x)f(f(x)) = f(x)，就说这个操作是幂等的。

举几个例子：

操作验证是否幂等
绝对值abs(abs(-5)) = abs(5) = 5 = abs(-5)✅ 是
向下取整floor(floor(3.7)) = floor(3) = 3 = floor(3.7)✅ 是
平方根sqrt(sqrt(4)) = sqrt(2) ≠ sqrt(4)❌ 否

简单说：做一次和做多次，结果一样，就是幂等的。

Web 开发里的幂等性#

在 Web 开发里，幂等性的意思是：

无论你调用同一个操作多少次，对系统最终状态的影响，都和调用一次一样。

HTTP 方法里有明确的幂等性约定：

方法是否幂等原因
GET✅只读，不改变状态
PUT✅更新操作，多次更新同一个值，结果一样
DELETE✅删除一个已经不存在的东西，还是”不存在”
POST❌每次调用都可能创建新资源、产生新副作用

我们的简历押题接口是 POST，按规范来说是非幂等的。每次调用都会：

  1. 调用一次 AI（花钱）
  2. 生成一份新的结果
  3. 消耗用户的配额

我们的困境#

问题就在这里：

接口按规范设计是非幂等的，但现实场景要求它必须幂等。

用户的操作是不可预测的。你永远不知道他会做什么——网络慢多点了一次，鼠标坏了单击变双击，或者就是不小心。

这些情况没办法在前端完全拦截，所以必须在后端做保护。

解决方案是：给每个请求一个唯一的 ID（requestId），通过检查这个 ID，判断是不是重复请求。

如果是重复请求，直接返回之前的结果，不重新生成，不扣费。

二、幂等性检查的完整流程#

用一张流程图来看整个判断逻辑：

用户发送请求（携带 requestId）
         │
         ▼
  requestId 存在吗？
    │           │
   没有         有
    │           │
    ▼           ▼
 直接执行    查询数据库
             │
             ▼
    找到这个 requestId 的记录吗？
         │              │
        没有            有
         │              │
         ▼              ▼
      新请求         检查状态
      继续执行    ┌────┴────┬────────────┐
                 │         │            │
              SUCCESS   PENDING       FAILED
                 │         │            │
                 ▼         ▼            ▼
           ✅ 重复请求  ⏳ 处理中    允许重试
           直接返回    告诉用户      重新生成
           已有结果    稍后查询
           不扣费

四个关键设计原则：

  1. 用 requestId 作为请求的唯一标识，由前端生成 UUID，每次用户主动发起新请求时生成一个新的
  2. 只有 SUCCESS 状态才返回缓存结果——成功过的请求，重复来了直接返回，不扣费
  3. PENDING 状态拒绝重复——同一个请求还在处理中，告诉用户稍等，不要再起一个新的
  4. FAILED 状态允许重试——之前失败了，用户可以重新尝试（通常会用新的 requestId）

每种状态，都有它该走的路。不强求，不混淆，各归其位。

三、代码实现#

Step 1：在 DTO 里加入 requestId#

首先，在 ResumeQuizDto 里加一个可选的 requestId 字段：

src/interview/dto/resume-quiz.dto.ts
export class ResumeQuizDto {
  @ApiProperty({ description: '公司名称', example: '字节跳动', required: false })
  @IsString()
  @IsOptional()
  company?: string;


  @ApiProperty({ description: '岗位名称', example: '前端开发工程师' })
  @IsString()
  @IsNotEmpty()
  positionName: string;


  // ... 其他字段


  @ApiProperty({
    description: '请求ID（用于幂等性检查）',
    example: 'f47ac10b-58cc-4372-a567-0e02b2c3d479',
    required: false,
  })
  @IsUUID('4')
  @IsOptional()
  requestId?: string; // ← 幂等性检查的唯一标识，由前端生成
}

几个细节值得注意：

  • @IsUUID('4')：限定必须是 UUID v4 格式，防止传入奇怪的字符串
  • @IsOptional()：不强制要求，没有 requestId 的请求也能正常处理
  • 由前端生成：不是后端生成的——后端生成的话，每次请求都会是新 ID，幂等性就失去意义了

Step 2：消费记录里存 requestId#

为了支持幂等性检查，ConsumptionRecord 里需要把 requestId 存进去。放在 metadata 字段里：

// 消费记录的关键字段
{
  recordId: string;
  userId: string;
  type: ConsumptionType;
  status: ConsumptionStatus; // PENDING / SUCCESS / FAILED / REFUNDED


  metadata: {
    requestId: string;     // ⭐ 关键：用于幂等性检查
    promptVersion: string;
  };


  resultId: string;
  startedAt: Date;
  completedAt?: Date;
}

为什么放在 metadata 里而不是直接放顶层？

因为 requestId 是”请求元数据”，不是业务数据本身。放在 metadata 里，结构更清晰，也方便后续扩展（比如加上 clientIp、userAgent 等）。

Step 3：实现幂等性检查逻辑#

在 executeResumeQuiz 方法开始时，加入幂等性检查：

src/interview/services/interview.service.ts
private async executeResumeQuiz(
  userId: string,
  dto: ResumeQuizDto,
  progressSubject: Subject<ProgressEvent>,
): Promise<void> {
  let consumptionRecord: ConsumptionRecord | null = null;
  const recordId = uuidv4();
  const resultId = uuidv4();


  try {
    // ========== 幂等性检查 ==========
    if (dto.requestId) {
      const existingRecord = await this.consumptionRecordModel.findOne({
        userId,
        'metadata.requestId': dto.requestId, // ← MongoDB 嵌套字段用点符号
        status: {
          $in: [ConsumptionStatus.SUCCESS, ConsumptionStatus.PENDING],
        },
      });


      if (existingRecord) {
        if (existingRecord.status === ConsumptionStatus.SUCCESS) {
          // ✅ 重复请求，且之前已经成功——直接返回缓存，不扣费
          const existingResult = await this.quizResultModel.findOne({
            resultId: existingRecord.resultId,
          });


          progressSubject.next({
            type: 'complete',
            progress: 100,
            label: 'AI 已完成问题生成',
            message: 'AI 已完成问题生成',
            stage: 'done',
            data: {
              questions: existingResult?.questions ?? [],
              summary: existingResult?.summary ?? '',
              remainingCount: await this.getRemainingCount(userId, 'resume'),
              isFromCache: true, // ⭐ 标记这是缓存结果
            },
          });
          progressSubject.complete();
          return;
        }


        if (existingRecord.status === ConsumptionStatus.PENDING) {
          // ⏳ 同一个请求还在处理中
          throw new BadRequestException('请求正在处理中，请稍后查询结果');
        }
      }
    }


    // ========== 正常生成流程 ==========
    this.logger.log(`✅ 用户扣费成功`);


    consumptionRecord = await this.consumptionRecordModel.create({
      recordId,
      user: new Types.ObjectId(userId),
      userId,
      type: ConsumptionType.RESUME_QUIZ,
      status: ConsumptionStatus.PENDING,
      consumedCount: 1,
      description: `简历押题 - ${dto?.company} ${dto.positionName}`,
      inputData: {
        company: dto?.company || '',
        positionName: dto.positionName,
        minSalary: dto.minSalary,
        maxSalary: dto.maxSalary,
        jd: dto.jd,
        resumeId: dto.resumeId,
      },
      resultId,
      metadata: {
        requestId: dto.requestId, // ← 存入 requestId，供下次幂等性检查使用
        promptVersion: dto.promptVersion,
      },
      startedAt: new Date(),
    });


    // 步骤 3：调用 AI 生成题目（TODO：后续实现）
    // ...


  } catch (error) {
    if (consumptionRecord) {
      await this.consumptionRecordModel.updateOne(
        { _id: consumptionRecord._id },
        { status: ConsumptionStatus.FAILED, error: error.message },
      );
      await this.quotaService.refund(userId, 1);
    }


    progressSubject.error(error);
  }
}

这里有一个 MongoDB 查询语法的细节，值得单独说一下：

// ❌ 错误写法——这样查不到嵌套字段
{ requestId: dto.requestId }


// ✅ 正确写法——嵌套字段用"点符号"（dot notation）
{ 'metadata.requestId': dto.requestId }

MongoDB 里查询嵌套对象的字段，必须用点符号。这是一个很容易踩的坑，记住就好。

Step 4：注入 Schema#

src/interview/interview.module.ts
@Module({
  imports: [
    MongooseModule.forFeature([
      { name: ConsumptionRecord.name, schema: ConsumptionRecordSchema },
      { name: ResumeQuizResult.name, schema: ResumeQuizResultSchema },
      { name: User.name, schema: UserSchema },
    ]),
  ],
})
export class InterviewModule {}

四、测试：验证幂等性是否生效#

第一步：登录，获取 Token#

Terminal window
curl -X POST http://localhost:3000/user/login \
  -H "Content-Type: application/json" \
  -d '{
    "email": "test@example.com",
    "password": "123456"
  }'

第二步：发送押题请求#

关键：两次请求使用相同的 requestId。

Terminal window
curl -X POST http://localhost:3000/interview/resume/quiz/stream \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer 替换成你的token" \
  -d '{
    "requestId": "c39f5c2e-8611-4f06-a87c-14ffcb7149ec",
    "company": "阿里巴巴",
    "positionName": "前端开发工程师",
    "minSalary": 25,
    "maxSalary": 35,
    "jd": "熟练掌握Vue.js框架及生态...",
    "resumeContent": "我叫张三，10年前端开发经验..."
  }'

第三步：验证幂等性#

同时打开两个终端，执行同一条命令（requestId 完全相同）：

预期结果：

  • 第一个请求：正常走生成流程，最后 isFromCache: false
  • 第二个请求：如果第一个还在处理中（PENDING），收到报错 "请求正在处理中"；如果第一个已经完成（SUCCESS），直接收到缓存结果，isFromCache: true

五、顺便说说上一篇遗留的坑#

上一篇跑通了 SSE 流式推送，但有一个现象让我困惑了一会儿：

重启服务后，第一条进度消息明明是 progress: 5，但前端收到的第一条却是 progress: 10。

反复重启了好几次，结果都一样。

最后定位到原因，是一个很典型的 RxJS 热流（Hot Observable）时序问题。

问题复现#

generateResumeQuizWithProgress 方法的原始写法：

generateResumeQuizWithProgress(userId, dto) {
  const subject = new Subject<ProgressEvent>()


  // ⚠️ 问题在这里：立刻执行，不等订阅建立
  void this.executeResumeQuiz(userId, dto, subject)


  return subject
}

执行顺序是这样的：

1. 进入 generateResumeQuizWithProgress()
2. 创建 subject
3. 立刻执行 executeResumeQuiz()
4. 第一轮循环：subject.next(5%)  ← 5% 在这里发出去了
5. generateResumeQuizWithProgress() 才 return subject
6. Controller 拿到 subject，才开始 .subscribe(...)  ← 订阅在这里才建立

问题就在第 4 步和第 6 步之间：

Subject 是”热的”（Hot Observable）——它不会帮你缓存历史消息。谁订阅得晚，谁就错过了之前的事件。

所以第一条 5% 在订阅建立之前就已经发出去了，客户端自然从第二条 10% 开始收到。

修复成功后，进度正常了
修复成功后，进度正常了

修复方式#

用 queueMicrotask 把”开始执行推送”推迟到当前调用栈结束之后：

generateResumeQuizWithProgress(userId, dto) {
  const subject = new Subject<ProgressEvent>()


  // ✅ 修复：等当前同步调用栈走完，让订阅先建立，再开始推送
  queueMicrotask(() => {
    void this.executeResumeQuiz(userId, dto, subject)
  })


  return subject
}

执行顺序变成了：

1. 进入 generateResumeQuizWithProgress()
2. 创建 subject
3. 把 executeResumeQuiz 注册到 microtask 队列（还没执行）
4. return subject
5. Controller 建立 .subscribe(...)  ← 订阅先建立
6. 当前调用栈结束
7. microtask 执行：executeResumeQuiz 开始运行
8. subject.next(5%)  ← 这时候订阅已经建立了，5% 能被收到

修复之后，前端收到的第一条就是 progress: 5 了。

这里学到的两个点#

第一：Subject 是热流，不缓存历史消息。

如果你需要”晚订阅的人也能收到之前的消息”，可以用 ReplaySubject(n)，它会缓存最近 n 条消息。但这个场景不需要，用 queueMicrotask 保证时序就够了。

第二：void someAsyncFn() 是”故意忽略 Promise 返回值”的写法。

如果直接写：

this.executeResumeQuiz(userId, dto, subject)

编辑器会提示：这个 Promise 没有被 await，也没有 .catch()，是一个”floating promise”，可能会漏掉错误。

加上 void 的意思是：

“我知道它返回 Promise，但这里就是要 fire-and-forget，请不要再提示我漏处理返回值。”

注意：void 只是告诉编辑器”我故意不接这个 Promise”，它本身不处理错误。 真正的错误处理在 executeResumeQuiz 内部的 try/catch 里，出错后通过 progressSubject.error(...) 传给 Controller。

六、缓存结果 vs 原始结果#

幂等性检查通过后，从缓存返回的结果和原始生成的结果内容一样，但有一个关键区别：

// 原始生成的结果
{
  resultId: "...",
  questions: [...],
  summary: "...",
  remainingCount: 9,    // 扣了一次配额
  isFromCache: false,
}


// 从缓存返回的结果
{
  resultId: "...",
  questions: [...],
  summary: "...",
  remainingCount: 10,   // 配额没有变化
  isFromCache: true,
}

isFromCache: true 意味着：这次请求没有扣费。

前端可以利用这个字段做 UI 处理，比如不显示”已消耗 1 次配额”的提示。

七、四个真实场景#

把上面的逻辑用具体场景串一遍，更容易记住：

场景 1：用户第一次请求（正常流程）

前端生成 requestId = "uuid-A"
发送请求
↓
后端查询：没找到 requestId = "uuid-A" 的记录
↓
执行完整生成流程，扣 1 次配额
↓
创建消费记录，status = SUCCESS，metadata.requestId = "uuid-A"
↓
返回结果，isFromCache: false

场景 2：用户重复请求（幂等性保护）

前端再次发送 requestId = "uuid-A"（与第一次相同）
↓
后端查询：找到 requestId = "uuid-A"，status = SUCCESS
↓
直接返回之前的结果，不扣费，不调用 AI
↓
返回结果，isFromCache: true

场景 3：请求还在处理中

用户发送 requestId = "uuid-A"
↓
后端正在生成（需要 10 秒），status = PENDING
↓
用户不耐烦，再发一次（仍然是 requestId = "uuid-A"）
↓
后端查询：找到记录，status = PENDING
↓
拒绝请求，返回错误："请求正在处理中，请稍后查询结果"

场景 4：用户主动重新生成

第一次：requestId = "uuid-A" → 成功
↓
用户点"重新生成"按钮
↓
前端生成新的 requestId = "uuid-B"
↓
后端查询：没找到 requestId = "uuid-B" 的记录
↓
执行完整生成流程，扣 1 次配额

关键在于：用户主动重新生成时，前端要生成一个新的 requestId。 这样后端就知道这是一个新请求，而不是重复请求。

八、常见问题#

Q1：为什么查询时用 $in: [SUCCESS, PENDING]，不查 FAILED？#

  • SUCCESS：之前成功过，现在是重复请求，直接返回缓存
  • PENDING：同一个请求还在处理中，不应该再起一个新的
  • FAILED：之前失败了，用户可能想重试，应该允许重新生成
  • REFUNDED：已退款的历史记录，不影响新请求

所以只查 SUCCESS 和 PENDING，FAILED 和 REFUNDED 的记录不拦截。

Q2：幂等性检查要不要设置过期时间？#

可以加，也可以不加，取决于业务需求。

如果加了 24 小时过期：

const existingRecord = await this.consumptionRecordModel.findOne({
  userId,
  'metadata.requestId': dto.requestId,
  status: { $in: [ConsumptionStatus.SUCCESS, ConsumptionStatus.PENDING] },
  createdAt: {
    $gte: new Date(Date.now() - 24 * 60 * 60 * 1000),
  },
});

对于简历押题这个场景，我倾向于不加过期时间——用户想重新生成，就应该明确点”重新生成”按钮，前端生成新的 requestId。

边界清晰，意图明确，不靠时间来模糊处理。

Q3：前端怎样生成 requestId？#

import { v4 as uuidv4 } from 'uuid';


// 用户点击"生成题目"按钮时
const handleGenerate = async () => {
  const requestId = uuidv4(); // 每次点击都生成一个新的 UUID


  const response = await fetch('/interview/resume/quiz/stream', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`,
    },
    body: JSON.stringify({
      positionName: '前端开发工程师',
      jd: '...',
      resumeContent: '...',
      requestId,
    }),
  });
};


// 用户点击"重新生成"按钮时
const handleRegenerate = async () => {
  const requestId = uuidv4(); // 重新生成一个新的 UUID，触发新的生成流程
  // ...
};

关键原则：每次用户主动发起新请求，就生成一个新的 requestId；重试同一个请求，就用同一个 requestId。

总结#

这篇文章，我们做了一件事：让接口不怕用户乱点。

回顾一下学到的东西：

  • ✅ 幂等性是什么：同一个操作执行多次，结果和执行一次一样
  • ✅ 为什么需要它：防止重复生成，节省成本，保护用户配额
  • ✅ 怎样实现：用 requestId 作为唯一标识，查数据库判断是否重复
  • ✅ 状态机设计：PENDING 拒绝重复，SUCCESS 返回缓存，FAILED 允许重试
  • ✅ MongoDB 嵌套查询：用点符号 'metadata.requestId' 查嵌套字段
  • ✅ 顺带搞清楚了两个坑：Subject 热流的时序问题，以及 void 的真实含义

幂等性检查是生产级系统的标配。 它不只是省钱，更是让系统在面对不可预测的用户行为时，依然能稳定运转。

每个请求都有自己的 ID，每个 ID 都有自己的状态。 该处理的处理，该拒绝的拒绝，该返回缓存的返回缓存。

各归其位，系统才稳。

这其实和做任何事情的道理一样—— 不是靠强力控制每一个变量， 而是把边界划清楚，让每个部分按自己的轨道走。

下一篇，我们把”真实 AI 内容生成”接进来——从”假进度 + 空结果”升级成”真实调用 DeepSeek 生成题目 + 返回完整结果”。

Prompt 怎么设计、怎样解析 AI 输出、怎样保证输出格式稳定……这些都在下一篇。

昇哥 · 2026年3月 90后 JS 全栈 × AI 学习途中，把踩过的坑写下来 专注羽毛球，爱音乐，正在研究易经 🎵🏸

支持与分享

如果这篇文章对你有帮助，欢迎分享给更多人或赞助支持！

赞助
【自学AI】12 幂等性：让你的接口不怕用户乱点
https://blog.fridolph.top/posts/2026-02-27__ai-idempotency/
作者
Fridolph
发布于
2026-03-16
许可协议
CC BY-NC-SA 4.0
「JS全栈AI Agent学习」八、A2A 协议完全指南：从零理解 Agent 协作体系
「JS全栈AI Agent学习」四、MCP：给AI工具世界造一个USB接口
相关文章 智能推荐
1
【自学AI】11 在写代码之前，先把设计想清楚
AI 2026-02-20
2
【自学AI】08 LangChain.js 实战：用 30 行代码让 AI 分析一份简历
AI 2026-02-15
3
【自学AI】09 让 AI 记住你说过的话——多轮对话与会话管理实战
AI 2026-02-16
4
「JS全栈AI Agent学习」二、反思、工具使用、规划——让 Agent 从"执行者"变成"自主完成者"
AI 2026-03-02
5
「JS全栈 AI Agent 学习」三、从独奏到乐队：多智能体、记忆管理与学习适应
AI 2026-03-23
随机文章 随机推荐

评论区

分享你的想法，与大家交流讨论

Profile Image of the Author
Fridolph
热爱 Coding、音乐和羽毛球的 90 后全栈工程师
公告
欢迎访问我的小站 ^_^ 我是昇哥，热爱Coding，喜爱音乐、羽毛球和摄影的 90后全栈工程师
分类
标签
© 2026 Fridolph. All Rights Reserved. / RSS / Sitemap
Powered by Astro & Firefly

文章目录