Express如何处理同一用户重复异步请求？如何实现串行处理？

核心疑问解答 #

• Express本身是异步非阻塞的，不会自动逐个处理同一用户的请求。当用户1秒内发5次请求时，这5个请求会被Express同时接收并启动处理流程——此时每个请求都会先去数据库查询最后更新时间，而因为耗时任务还没执行完（计数器尚未更新），这5次查询都会得到“距上次更新超过24小时”的结果，最终导致函数执行5次、计数器被递增5次，完全符合你担心的数据库更新未同步问题。

解决方案：实现同一用户请求的串行化 #

要避免这个问题，需要为每个用户维护请求控制机制，确保同一用户的请求只能串行执行，前一个请求处理完成后再处理下一个。以下是几种可行的实现方式：

1. 内存级请求队列（单实例场景） #

在服务器内存中维护一个以用户ID为键的Map，每个键对应一个Promise队列，用来控制同一用户的请求串行：

const userRequestQueues = new Map();

async function processUserRequest(userId, handler) {
  // 如果用户没有队列，初始化一个已resolved的Promise作为起点
  if (!userRequestQueues.has(userId)) {
    userRequestQueues.set(userId, Promise.resolve());
  }

  // 将当前请求加入队列，等待前一个请求完成
  const queue = userRequestQueues.get(userId);
  const newQueue = queue.then(async () => {
    try {
      return await handler();
    } finally {
      // 队列处理完成后，如果没有后续请求，清理内存（可选）
      if (userRequestQueues.get(userId) === newQueue) {
        userRequestQueues.delete(userId);
      }
    }
  });

  userRequestQueues.set(userId, newQueue);
  return newQueue;
}

// 在Express路由中使用
app.post('/your-endpoint', async (req, res) => {
  const userId = req.user.id; // 假设从请求中获取用户ID
  try {
    await processUserRequest(userId, async () => {
      // 业务逻辑：检查最后更新时间 -> 执行耗时任务 -> 更新计数器
      const user = await db.User.findById(userId);
      const now = Date.now();
      if (now - user.lastUpdateTime < 24 * 60 * 60 * 1000) {
        return res.status(403).send('需等待24小时后再请求');
      }

      // 执行耗时20-30秒的异步任务
      await longRunningAsyncTask();

      // 更新计数器和最后更新时间
      await db.User.findByIdAndUpdate(userId, {
        $inc: { counter: 1 },
        lastUpdateTime: now
      });

      res.send('任务执行完成');
    });
  } catch (err) {
    res.status(500).send('服务器错误');
  }
});

2. 分布式锁（多实例/集群场景） #

如果你的Express应用是多实例部署的，内存级队列无法跨实例生效，这时需要用分布式锁来确保同一用户的请求串行执行：

• 基于Redis的SETNX命令实现分布式锁：在处理请求前，先尝试获取以用户ID为键的锁，只有获取到锁的请求才能继续执行，其他请求要么排队等待，要么直接返回“请求处理中”。
• 示例逻辑：

  const redis = require('redis');
  const client = redis.createClient();
  
  async function acquireLock(userId) {
    const lockKey = `lock:user:${userId}`;
    // 设置锁，过期时间设为比耗时任务最长时间多一点（比如40秒），避免死锁
    const result = await client.set(lockKey, 'locked', { NX: true, EX: 40 });
    return result === 'OK';
  }
  
  async function releaseLock(userId) {
    const lockKey = `lock:user:${userId}`;
    await client.del(lockKey);
  }
  
  // 路由中使用
  app.post('/your-endpoint', async (req, res) => {
    const userId = req.user.id;
    let lockAcquired = false;
    try {
      lockAcquired = await acquireLock(userId);
      if (!lockAcquired) {
        return res.status(429).send('请求正在处理中，请稍后再试');
      }
  
      // 业务逻辑：检查时间、执行任务、更新计数器（同内存队列版）
      // ...
  
    } catch (err) {
      res.status(500).send('服务器错误');
    } finally {
      if (lockAcquired) {
        await releaseLock(userId);
      }
    }
  });
  

3. 数据库事务+行级锁（兜底方案） #

如果不想引入额外组件，也可以利用数据库的行级锁来避免并发问题：

• 在查询用户数据时，使用数据库的行锁机制（比如MySQL的SELECT ... FOR UPDATE，MongoDB的findOneAndUpdate带原子操作），确保同一时间只有一个请求能修改该用户的记录：

  // MongoDB示例：原子性检查并更新
  app.post('/your-endpoint', async (req, res) => {
    const userId = req.user.id;
    const now = Date.now();
    const twentyFourHours = 24 * 60 * 60 * 1000;
  
    try {
      // 原子性检查最后更新时间，并仅当符合条件时更新
      const updatedUser = await db.User.findOneAndUpdate(
        {
          _id: userId,
          lastUpdateTime: { $lt: now - twentyFourHours }
        },
        {
          $inc: { counter: 1 },
          lastUpdateTime: now
        },
        { new: true }
      );
  
      if (!updatedUser) {
        return res.status(403).send('需等待24小时后再请求');
      }
  
      // 执行耗时任务（注意：这里如果任务失败，计数器已经更新了，需要根据业务决定是否回滚）
      await longRunningAsyncTask();
  
      res.send('任务执行完成');
    } catch (err) {
      res.status(500).send('服务器错误');
    }
  });
  

  注意：这种方式下，即使多个请求同时到达，数据库也只会允许一个请求成功更新记录，其他请求会返回updatedUser为null，直接拒绝。但如果你的耗时任务必须在计数器更新前执行，这种方式需要调整逻辑（比如先锁用户记录，执行任务后再更新）。

补充说明 #

• 如果选择串行化队列，对于同一用户的后续请求，你可以选择让请求排队等待，或者直接返回“请求处理中”的提示，取决于你的业务需求。
• 多实例场景下，内存队列无法共享，必须用分布式锁或者数据库锁来保证全局的串行性。

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