Node.js+Mongoose查询MongoDB性能优化及分页需求解决

我有一个基于Node.js和MongoDB的API，用Mongoose做ODM，MongoDB集合名为Battery，存储多IoT设备的数据：

• 每个文档里的BAT数组只包含一个对象
• 现有约10万条文档，每4分钟新增一条
• 存在多个如"S044"、"S067"的SID值，唯一SID数量在10到2000之间
• 每个SID对应的BID范围是1到「该SID最新文档的BYC/BXC计算值」（每个SID的BYC和BXC是唯一值）

原查询代码响应耗时约8秒，而且用数组存储结果没法实现分页，需要优化查询/聚合逻辑。

原代码 #

const getLatestBats = async (all_SIDs) => {
  const distinctSIDs = await Battery.distinct("SID", { SID: { $in: all_SIDs } });

  const latestDocuments = [];

  for (const SID of distinctSIDs) {
    const batteryDoc = await BAT.findOne({ SID })
      .select("BAT")
      .sort({ TS: -1 });

    if (batteryDoc && batteryDoc.BAT && batteryDoc.BAT.length > 0) {
      const { BYC, BXC } = batteryDoc.BAT[0];
      const BatCount = BYC / BXC;

      const documents = await Battery.aggregate([
        { $match: { SID, "BAT.BID": { $in: [...Array(BatCount)].map((_, index) => index + 1) } } },
        { $sort: { TS: -1 } },
        { $group: { _id: "$BAT.BID", document: { $first: "$$ROOT" } } },
        { $replaceRoot: { newRoot: "$document" } },
        { $project: { _id: 0, __v: 0 } },
     
      ]);

      latestDocuments.push(...documents);
    }
  }

  return { data: latestDocuments};
};


router.get('/', async (req, res) => {
  try {    
// all_SIDs通过中间件获取
  const batteries = await getLatestBats(all_SIDs);
    res.json(batteries);

} catch (error) {
  console.error(error);
  res.status(500).json({ message: 'Server error' });
  }
});

module.exports = router;

集合文档示例 #

_id: 6483ea992dae73e915e14575
TS: "1686366869373"
SID: "CM045"
BAT: [
  {
    BID: 1,
    BXC: 100,
    BYC: 500
  }
]
__v: 0

_id: 6483ea9f2dae73e915e14578
TS: "1686366879348"
SID: "CM045"
BAT: [
  {
    BID: 4,
    BXC: 100,
    BYC: 500
  }
]
__v: 0

1. 先加索引（核心优化） #

原代码性能差的关键原因之一是缺少合适索引，每次查询都要全表扫描。创建以下复合索引：

// 在Mongoose的Battery模型中添加
Battery.index({ SID: 1, TS: -1, "BAT.BID": 1 });

额外提示：当前TS是字符串类型，排序和比较效率低，建议改成Number类型存储时间戳数字。如果暂时无法修改数据结构，聚合时要先转成数字处理。

2. 重构聚合逻辑，消除循环查询 #

把原来的N次查询改成单次聚合，大幅减少数据库请求次数：

• 按SID分组，获取每个SID的最新文档（最大TS）并计算BatCount
• 关联原集合，筛选每个SID下BID在1~BatCount范围内的文档
• 按SID+BID分组，取每个BID对应的最新文档
• 处理投影和分页逻辑

3. 实现分页 #

在聚合末尾添加$skip和$limit实现基础分页，若数据量持续增长，可改用基于TS的游标分页（避免$skip的性能问题）。

const getLatestBats = async (all_SIDs, page = 1, limit = 20) => {
  const skip = (page - 1) * limit;

  const pipeline = [
    // 筛选目标SID，按SID分组取最新文档
    { $match: { SID: { $in: all_SIDs } } },
    { $addFields: { tsNumber: { $toLong: "$TS" } } }, // 字符串TS转数字，提升排序效率
    { $sort: { SID: 1, tsNumber: -1 } },
    {
      $group: {
        _id: "$SID",
        BYC: { $first: "$BAT.0.BYC" },
        BXC: { $first: "$BAT.0.BXC" }
      }
    },
    // 计算每个SID的BID上限
    { $addFields: { maxBid: { $divide: ["$BYC", "$BXC"] } } },
    // 关联原集合，获取符合BID范围的文档
    {
      $lookup: {
        from: "batteries", // 注意这里是MongoDB集合的实际名称（默认小写复数）
        let: { sid: "$_id", bidLimit: "$maxBid" },
        pipeline: [
          { $match: { $expr: { $and: [{ $eq: ["$SID", "$$sid"] }, { $lte: ["$BAT.0.BID", "$$bidLimit"] }] } } },
          { $addFields: { tsNumber: { $toLong: "$TS" } } },
          { $sort: { tsNumber: -1 } }
        ],
        as: "sidDocuments"
      }
    },
    // 展开关联结果
    { $unwind: "$sidDocuments" },
    // 按SID+BID分组，取每个BID的最新文档
    {
      $group: {
        _id: { sid: "$_id", bid: "$sidDocuments.BAT.0.BID" },
        latestDoc: { $first: "$sidDocuments" }
      }
    },
    // 替换根节点并清理字段
    { $replaceRoot: { newRoot: "$latestDoc" } },
    { $project: { _id: 0, __v: 0, tsNumber: 0 } },
    // 分页处理
    { $skip: skip },
    { $limit: limit }
  ];

  const data = await Battery.aggregate(pipeline);
  // 计算总条数，用于分页导航
  const total = await Battery.aggregate([
    ...pipeline.slice(0, -2), // 移除skip和limit
    { $count: "total" }
  ]).then(res => res[0]?.total || 0);

  return { data, total, page: parseInt(page), limit: parseInt(limit) };
};

router.get('/', async (req, res) => {
  try {
    // 从请求参数获取分页配置，默认第一页每页20条
    const { page = 1, limit = 20 } = req.query;
    // all_SIDs通过中间件获取
    const batteries = await getLatestBats(all_SIDs, parseInt(page), parseInt(limit));
    res.json(batteries);
  } catch (error) {
    console.error(error);
    res.status(500).json({ message: 'Server error' });
  }
});

module.exports = router;

• 如果BYC和BXC是每个SID的固定值，建议单独建一张Device集合存储，避免每次聚合都计算
• 定期归档/删除旧数据：如果只需要每个BID的最新文档，历史数据可以清理，减少集合大小
• 用await Battery.aggregate(pipeline).explain()分析聚合查询，确认索引被正确使用

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