我之前做电商系统的时候就遇到过一模一样的场景——同一用户的操作必须严格串行执行，不然很容易出现数据不一致的问题（比如添加商品的数据库事务还没提交，清空购物车的操作就已经把数据删了，结果逻辑完全乱掉）。而不同用户之间的操作完全可以并行，互不影响。用单用户专属任务队列是最直接也最可靠的解决方案，我来拆解具体怎么落地：

核心思路：为每个用户维护独立队列 #

给每个用户ID（或者会话ID，如果是无状态服务的话）创建一个专属的任务队列。当用户发起新操作时，不是立即执行，而是把整个操作的"完整执行逻辑包"塞进对应的队列里。然后由一个消费进程/线程，按顺序逐个取出队列里的任务执行——必须等上一个任务的所有阶段（调用外部REST API、查询数据库、执行业务JavaScript代码）全部完成后，才会启动下一个任务。

任务定义：封装完整操作流程 #

每个任务应该是一个包含全流程逻辑的异步函数（毕竟涉及API调用、数据库操作这些IO密集型操作，用异步能最大化服务性能）。举个实际的例子：

// 封装「添加商品到购物车」的完整任务
async function addItemTask(userId, itemData) {
  try {
    // 阶段1：调用商品服务API验证商品合法性
    const validItem = await fetch('/api/products/validate', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(itemData)
    }).then(res => res.json());

    // 阶段2：查询用户当前购物车信息
    const currentCart = await db.query('SELECT id FROM carts WHERE user_id = ?', [userId]);

    // 阶段3：执行添加逻辑并更新数据库
    await db.query(
      'INSERT INTO cart_items (cart_id, product_id, quantity) VALUES (?, ?, ?)',
      [currentCart.id, validItem.id, itemData.quantity]
    );

    // 可选：更新购物车总价等后续操作
    await updateCartTotal(currentCart.id);
  } catch (err) {
    console.error(`用户${userId}添加商品任务失败：`, err);
    // 这里可以根据业务决定是否重试，或者通知用户
  }
}

// 封装「清空购物车」的完整任务
async function clearCartTask(userId) {
  try {
    // 阶段1：查询用户购物车ID
    const cart = await db.query('SELECT id FROM carts WHERE user_id = ?', [userId]);
    if (!cart) return;

    // 阶段2：删除购物车所有商品
    await db.query('DELETE FROM cart_items WHERE cart_id = ?', [cart.id]);

    // 阶段3：更新购物车状态
    await db.query('UPDATE carts SET total_items = 0, total_amount = 0 WHERE id = ?', [cart.id]);
  } catch (err) {
    console.error(`用户${userId}清空购物车任务失败：`, err);
  }
}

队列管理：内存或持久化存储按需选择 #

• 单实例服务场景：可以用内存对象来存储用户队列，比如用Map来关联用户ID和队列。推荐用成熟的异步队列库（比如p-queue）来简化实现，不用自己写队列的消费逻辑：

  const PQueue = require('p-queue');
  const userTaskQueues = new Map();
  
  // 获取或创建用户专属队列（并发数设为1，强制串行）
  function getUserQueue(userId) {
    if (!userTaskQueues.has(userId)) {
      const queue = new PQueue({ concurrency: 1 });
      userTaskQueues.set(userId, queue);
      // 可选优化：用户1小时无操作后自动清理空队列，避免内存泄漏
      setTimeout(() => {
        if (queue.size === 0 && queue.pending === 0) {
          userTaskQueues.delete(userId);
        }
      }, 3600000);
    }
    return userTaskQueues.get(userId);
  }
  
  // 处理用户操作的入口函数
  async function handleUserRequest(userId, operationType, operationData) {
    const queue = getUserQueue(userId);
    let task;
  
    switch(operationType) {
      case 'addItem':
        task = () => addItemTask(userId, operationData);
        break;
      case 'clearCart':
        task = () => clearCartTask(userId);
        break;
      // 扩展其他操作类型...
    }
  
    // 将任务加入队列，自动按顺序执行
    await queue.add(task);
    return { code: 200, message: '操作已完成' };
  }
  

• 多实例分布式场景：内存队列就不够用了，得用持久化消息队列。比如用Redis的List结构，给每个用户创建一个名为user-task-queue:{userId}的队列，生产者把任务序列化后推入对应队列，消费者监听这些队列并串行执行任务。如果用RabbitMQ，可以给每个用户创建专属队列，或者用路由键来区分用户，保证同一用户的任务被同一个消费者处理。

关键注意事项 #

• 错误隔离：每个任务内部一定要加异常捕获，避免单个任务失败导致整个队列阻塞。比如上面示例中的try/catch，即使某个任务失败，队列也会继续执行下一个任务。
• 任务幂等性：网络波动可能导致用户重复提交操作，所以每个任务要保证幂等——比如添加商品时，先检查商品是否已经在购物车里，避免重复插入；清空购物车时，即使执行多次，结果也是一样的。
• 资源清理：不管是内存队列还是持久化队列，都要注意清理长时间无活动的用户队列，避免资源浪费。比如内存队列的超时清理，Redis队列可以设置过期时间。

内容的提问来源于stack exchange，提问作者Đinh Anh Huy