Express.js+MongoDB高流量场景下解决库存超卖（Race Condition）问题

我用Express.js和MongoDB搭建了带闪购功能的电商网站，闪购时段每分钟有5000名用户访问，目前遇到高流量下库存负数（超卖）的问题，推测是竞态条件导致：当库存仅剩1件时，多名用户同时查询到库存充足，完成交易后库存被多次扣减，最终出现负数（比如1-5=-4）。

最初的代码：

const product = await this.productService.findOne(slug)

if( !product )
  return res.status(404).json({ success: false, message: 'Not Found' })

if( product.available <= 0 )
  return res.json({ success: false, message: 'Out of Stock' })

const newTransaction = await this.transactionService.create(user, product)

await this.productService.update(product.id, { available: product.available - 1 })

return res.json({ success: true, data: newTransaction })

之后添加了随机延迟临时缓解，当前代码：

await sleep(200, 500) // 自定义随机延迟函数，参数为最小/最大毫秒数

const product = await this.productService.findOne(slug)

if( !product )
  return res.status(404).json({ success: false, message: 'Not Found' })

if( product.available <= 0 )
  return res.json({ success: false, message: 'Out of Stock' })

const newTransaction = await this.transactionService.create(user, product)

await this.productService.update(product.id, { available: product.available - 1 })

return res.json({ success: true, data: newTransaction })

当前代码能减少80-90%的库存负数，但极高流量下仍会超卖，且本地无法复现（需要同时发2000-5000次请求），求可行解决方案及高并发模拟工具。

1. 使用MongoDB原子更新操作 #

核心是把「查询库存+扣减库存」合并为一个原子操作，避免竞态。直接在update时判断库存是否大于0，同时完成扣减：

// 替换原有的查询+更新逻辑
const updatedProduct = await this.productService.updateOne(
  { slug, available: { $gt: 0 } }, // 查询条件：商品存在且库存>0
  { $inc: { available: -1 } } // 原子扣减库存
)

// 检查是否有文档被修改
if (updatedProduct.modifiedCount === 0) {
  return res.json({ success: false, message: 'Out of Stock' })
}

// 库存扣减成功后再创建交易记录
const newTransaction = await this.transactionService.create(user, updatedProduct)

return res.json({ success: true, data: newTransaction })

updateOne是原子操作，MongoDB会保证同一时间只有一个请求能满足条件并完成扣减，彻底避免竞态。

2. 乐观锁（版本号机制） #

给商品文档加一个version字段，每次更新时校验版本号，只有版本匹配才允许更新：

// 1. 查询商品及版本号
const product = await this.productService.findOne(slug)
if (!product) return res.status(404).json({ success: false, message: 'Not Found' })
if (product.available <= 0) return res.json({ success: false, message: 'Out of Stock' })

// 2. 尝试原子更新：版本号匹配且库存>0时扣减
const updatedProduct = await this.productService.updateOne(
  { _id: product._id, version: product.version, available: { $gt: 0 } },
  { $inc: { available: -1 }, $set: { version: product.version + 1 } }
)

if (updatedProduct.modifiedCount === 0) {
  // 版本不匹配或库存不足，重试或返回失败
  return res.json({ success: false, message: 'Transaction failed, please try again' })
}

// 3. 创建交易记录
const newTransaction = await this.transactionService.create(user, product)
return res.json({ success: true, data: newTransaction })

如果并发请求导致版本不匹配，需要前端引导用户重试，适合对数据一致性要求高的场景。

3. 悲观锁（MongoDB事务+findOneAndUpdate） #

用MongoDB的事务结合findOneAndUpdate的排他锁，锁定商品文档直到操作完成：

const session = await this.productService.startSession()
session.startTransaction()

try {
  // 查询并锁定商品，直到事务提交
  const product = await this.productService.findOneAndUpdate(
    { slug, available: { $gt: 0 } },
    { $inc: { available: -1 } },
    { session, returnOriginal: false } // 返回更新后的文档
  )

  if (!product) {
    await session.abortTransaction()
    return res.json({ success: false, message: 'Out of Stock' })
  }

  // 在同一个事务中创建交易记录
  await this.transactionService.create(user, product, { session })

  await session.commitTransaction()
  return res.json({ success: true, data: product })
} catch (err) {
  await session.abortTransaction()
  return res.status(500).json({ success: false, message: 'Server Error' })
} finally {
  session.endSession()
}

注意MongoDB事务需要副本集或分片集群，单节点不支持，适合需要跨文档原子操作的场景。

4. 消息队列削峰 #

把下单请求放入消息队列（比如Redis Queue、BullMQ），后台用单进程或有限进程消费，串行处理库存扣减：

// 前端请求接口：将下单请求加入队列
const job = await this.queue.add('place-order', { userId: user.id, productSlug: slug })

return res.json({ success: true, message: 'Order processing, please wait' })

// 后台消费队列
this.queue.process('place-order', async (job) => {
  const { userId, productSlug } = job.data
  const user = await this.userService.findOne(userId)
  const product = await this.productService.findOne(productSlug)

  if (!product || product.available <= 0) {
    await this.notificationService.send(user, 'Order failed: Out of Stock')
    return
  }

  await this.productService.updateOne({ _id: product._id }, { $inc: { available: -1 } })
  const transaction = await this.transactionService.create(user, product)
  await this.notificationService.send(user, 'Order success', transaction)
})

这种方式能彻底解决并发问题，但会增加系统复杂度，适合流量极大的闪购场景。

1. k6 #

轻量级的JavaScript压测工具，适合模拟高并发请求：
编写压测脚本load-test.js：

import http from 'k6/http'
import { check } from 'k6'

export const options = {
  vus: 5000, // 虚拟用户数
  duration: '1m', // 持续时间
}

export default function () {
  const res = http.post('http://your-api.com/api/order', JSON.stringify({
    slug: 'flash-sale-product',
    userId: 'test-user-id'
  }), {
    headers: { 'Content-Type': 'application/json' }
  })
  check(res, { 'status is 200': (r) => r.status === 200 })
}

运行命令：k6 run load-test.js

2. Artillery #

基于Node.js的压测工具，配置简单：
编写config.yml：

config:
  target: 'http://your-api.com'
  phases:
    - duration: 60
      arrivalRate: 5000 # 每分钟5000请求
scenarios:
  - flow:
      - post:
          url: '/api/order'
          json:
            slug: 'flash-sale-product'
            userId: 'test-user-id'

运行命令：artillery run config.yml

3. Locust #

Python编写的分布式压测工具，支持Web UI监控：
编写locustfile.py：

from locust import HttpUser, task, between

class OrderUser(HttpUser):
    wait_time = between(0, 0) # 无延迟，模拟并发

    @task
    def place_order(self):
        self.client.post("/api/order", json={
            "slug": "flash-sale-product",
            "userId": "test-user-id"
        })

运行命令：locust -f locustfile.py，然后访问http://localhost:8089设置并发数和请求速率。

内容的提问来源于stack exchange，提问作者Yovi Prasetyo