Node.js 操作 MongoDB

2026/11/13 · by ZHANG XING YUN · 阅读 2 次 · 2911 字 · 预计阅读 8 分钟

本文讲在 Node.js 里如何使用 MongoDB：驱动选型、连接与连接池、CRUD、游标、聚合、索引、事务、变更流、错误处理，以及 Mongoose 的使用方式。

环境：Node.js 20+、MongoDB 7、mongodb 6.x、mongoose 8.x
示例默认使用 CommonJS

一、驱动与选型

方案	特点	适用
`mongodb`（官方驱动）	贴近数据库、API 稳定、无额外抽象	需要完全控制、复杂聚合
`mongoose`	Schema、校验、中间件、`populate`	业务模型清晰、团队协作
`Prisma`	类型安全、统一 ORM（MongoDB 支持为预览特性）	已在用 Prisma 的项目

选择建议：先用官方驱动理解数据库行为；业务模型稳定、需要校验与关联时再上 Mongoose。两者可以在同一项目共存，但不要对同一集合混用两套写入逻辑。

二、安装与连接

npm i mongodb

// db.js
const { MongoClient } = require("mongodb");

const client = new MongoClient(process.env.MONGO_URL, {
	maxPoolSize: 10, // 连接池上限
	minPoolSize: 0,
	serverSelectionTimeoutMS: 5000, // 选主超时
	socketTimeoutMS: 45000,
	retryWrites: true, // 副本集下自动重试可重试的写操作
	appName: "my-api"
});

let db;

async function getDb() {
	if (!db) {
		await client.connect();
		db = client.db(process.env.MONGO_DB || "appdb");
	}
	return db;
}

module.exports = { client, getDb };

连接串：

mongodb://root:secret@localhost:27017/appdb?authSource=admin
mongodb+srv://user:pass@cluster0.example.mongodb.net/appdb?retryWrites=true&w=majority

带认证的容器必须写 authSource=admin，否则报 Authentication failed。

2.1 连接是惰性的

new MongoClient() 不会立即连接，第一次操作时才建立。因此建议在启动时主动 await client.connect()，让"连不上"在启动阶段暴露，而不是在第一个请求。

2.2 连接池

maxPoolSize：并发上限。默认 100，Node 服务通常 10~50 足够。
一个 Node 进程一个 MongoClient，不要每个请求 new 一个。
多副本部署时，总连接数 = 副本数 × maxPoolSize，需考虑数据库连接上限。

三、CRUD

const { ObjectId } = require("mongodb");
const { getDb } = require("./db");

async function createUser(input) {
	const db = await getDb();
	const result = await db.collection("users").insertOne({
		...input,
		createdAt: new Date()
	});
	return { id: result.insertedId, ...input };
}

async function findUser(id) {
	const db = await getDb();
	return db.collection("users").findOne({ _id: new ObjectId(id) });
}

async function listUsers(page = 1, size = 20) {
	const db = await getDb();
	return db
		.collection("users")
		.find({ status: "active" })
		.sort({ createdAt: -1 })
		.skip((page - 1) * size)
		.limit(size)
		.toArray();
}

async function updateUser(id, patch) {
	const db = await getDb();
	const result = await db
		.collection("users")
		.updateOne({ _id: new ObjectId(id) }, { $set: { ...patch, updatedAt: new Date() } });
	return result.matchedCount === 1;
}

async function deleteUser(id) {
	const db = await getDb();
	const result = await db.collection("users").deleteOne({ _id: new ObjectId(id) });
	return result.deletedCount === 1;
}

async function upsertByEmail(email, name) {
	const db = await getDb();
	const result = await db.collection("users").updateOne(
		{ email },
		{ $set: { name }, $setOnInsert: { createdAt: new Date() } },
		{ upsert: true }
	);
	return { upsertedId: result.upsertedId, matched: result.matchedCount };
}

返回值注意：

方法	返回
`insertOne`	`{ acknowledged, insertedId }`
`insertMany`	`{ acknowledged, insertedCount, insertedIds }`
`updateOne` / `updateMany`	`{ matchedCount, modifiedCount, upsertedId }`
`deleteOne` / `deleteMany`	`{ deletedCount }`
`findOneAndUpdate`	默认直接返回文档（6.x），不再包 `{ value }`

findOneAndUpdate 需要旧值时：

const before = await collection.findOneAndUpdate(
	{ _id },
	{ $inc: { views: 1 } },
	{ returnDocument: "before" } // "after" 返回更新后的文档
);

四、查询与游标

const db = await getDb();
const users = db.collection("users");

// 条件查询
await users.find({ status: "active", points: { $gte: 10 } }).toArray();

// 投影：只返回需要的字段
await users.find({}, { projection: { name: 1, email: 1, _id: 0 } }).toArray();

// 逻辑组合
await users.find({
	$or: [{ status: "active" }, { status: { $exists: false } }],
	email: { $regex: /@example\.com$/ }
}).toArray();

// 数组与嵌套
await users.find({ tags: "vip" }).toArray();
await users.find({ "profile.city": "上海" }).toArray();

4.1 游标：处理大结果集

// toArray 会把全部结果放进内存，数据量大时用游标逐条处理
const cursor = users.find({ status: "active" }).batchSize(500);
for await (const doc of cursor) {
	await handle(doc);
}

toArray() 适合小结果集；导出、批量处理、数据迁移用 for await 或 cursor.forEach。

4.2 分页

skip/limit 简单，但 skip 越大越慢。
深分页用游标（keyset）：find({ createdAt: { $lt: lastCreatedAt } }).sort({ createdAt: -1 }).limit(20)。

4.3 计数

await users.countDocuments({ status: "active" }); // 准确，会扫描
await users.estimatedDocumentCount(); // 用元数据，快但不精确

五、聚合

const db = await getDb();

const stats = await db.collection("orders").aggregate([
	{ $match: { status: "paid" } },
	{ $group: { _id: "$userId", total: { $sum: "$amount" }, count: { $sum: 1 } } },
	{ $sort: { total: -1 } },
	{ $limit: 10 }
]).toArray();

// 关联用户（$lookup 的 foreignField 必须有索引，且两边类型一致）
const withUser = await db.collection("orders").aggregate([
	{ $match: { status: "paid" } },
	{
		$lookup: {
			from: "users",
			localField: "userId",
			foreignField: "_id",
			as: "user"
		}
	},
	{ $unwind: { path: "$user", preserveNullAndEmptyArrays: true } },
	{ $project: { amount: 1, "user.name": 1 } }
]).toArray();

要点：

$match 放最前并命中索引，能显著减少后续阶段的数据量。
$lookup 是左连接，foreignField 没有索引时会全集合扫描。
聚合结果也要限制大小，必要时加 $limit。
allowDiskUse: true 允许聚合落盘，处理超大数据集时使用。

六、索引管理

await db.collection("users").createIndex({ email: 1 }, { unique: true });
await db.collection("orders").createIndex({ userId: 1, createdAt: -1 });
await db.collection("sessions").createIndex({ createdAt: 1 }, { expireAfterSeconds: 3600 });
await db.collection("users").createIndex(
	{ phone: 1 },
	{ unique: true, partialFilterExpression: { phone: { $exists: true } } }
);

// 查看与删除
await db.collection("users").indexes();
await db.collection("users").dropIndex("email_1");

生产建议：

索引在部署脚本或迁移里创建，不要依赖应用启动时自动建（多实例并发建索引会互相影响）。
用 explain("executionStats") 确认查询是否走索引：

const stats = await db.collection("orders")
	.find({ userId })
	.explain("executionStats");
console.log(stats.executionStats.totalDocsExamined, stats.executionStats.nReturned);

totalDocsExamined 远大于 nReturned 说明索引缺失或选择性差。

七、事务

事务要求副本集或分片集群（单机报 Transaction numbers are only allowed on a replica set member or mongos）。

const { client, getDb } = require("./db");

async function transfer(fromId, toId, amount) {
	const session = client.startSession();
	try {
		await session.withTransaction(async () => {
			const db = await getDb();
			const accounts = db.collection("accounts");

			await accounts.updateOne(
				{ _id: new ObjectId(fromId), balance: { $gte: amount } },
				{ $inc: { balance: -amount } },
				{ session }
			);
			await accounts.updateOne(
				{ _id: new ObjectId(toId) },
				{ $inc: { balance: amount } },
				{ session }
			);
		});
	} finally {
		await session.endSession();
	}
}

要点：

事务内所有操作都要带 { session }，漏一个就在事务外执行。
withTransaction 内置重试（可重试错误），比手写 startTransaction/commitTransaction 简单。
事务默认 60 秒超时，内部不要做远程调用。
优先用单文档原子操作，能不用事务就不用。

八、错误处理

try {
	await db.collection("users").insertOne({ email: "a@example.com" });
} catch (error) {
	if (error.code === 11000) {
		// 唯一索引冲突
		throw new Error("邮箱已存在");
	}
	throw error;
}

常见错误：

错误码 / 名称	含义	处理
`11000`	唯一键冲突	转为业务错误，或用 upsert
`MongoServerError: Authentication failed`	认证失败	检查 `authSource` 与账号密码
`MongoServerSelectionError`	连不上（网络、DNS、副本集）	检查连接串与网络
`MongoNetworkError`	网络中断	驱动会自动重试可重试操作；非幂等写要自己幂等
`BSONObjectTooLarge`	单文档超过 16MB	拆分文档
`DocumentValidationFailure` (121)	JSON Schema 校验失败	检查写入字段

重试原则：只重试幂等操作或可重试写（retryWrites: true + 副本集）。业务层用唯一索引 + upsert 保证幂等，比盲目重试更可靠。

九、ObjectId 与时间

const { ObjectId } = require("mongodb");

// 字符串转 ObjectId；非法字符串会抛错
new ObjectId("65f0c3a1e2b4c9d8f7a6b5c4");

// 校验，避免直接把用户输入传给 findById
ObjectId.isValid(idString);

// ObjectId 自带时间戳（创建时间，秒级）
new ObjectId(idString).getTimestamp();

常见错误是把字符串直接传给 find({ _id: id })：类型不匹配会查不到（返回 null），而不是报错。所有入参都要先转 ObjectId 并校验。

日期：

驱动把 Date 存为 BSON Date（UTC），读出来是 JS Date，时区转换在展示层做。
用 $set: { updatedAt: new Date() }，不要在应用里存字符串时间。
范围查询用 Date 对象：{ createdAt: { $gte: new Date("2026-01-01") } }。

十、变更流

const db = await getDb();
const changeStream = db.collection("orders").watch([
	{ $match: { "operationType": { $in: ["insert", "update"] } } }
]);

changeStream.on("change", change => {
	console.log(change.operationType, change.documentKey, change.fullDocument);
});

process.on("SIGTERM", async () => {
	await changeStream.close();
});

限制：

需要副本集或分片集群（单机不支持）。
变更流依赖 oplog，恢复窗口有限；断线后要用 resumeAfter 续传。
变更流适合事件通知（如刷新缓存、推送），不适合做可靠消息队列（需要更强投递保证时用 Kafka 等）。

十一、Mongoose 用法

npm i mongoose

const mongoose = require("mongoose");

await mongoose.connect(process.env.MONGO_URL, {
	maxPoolSize: 10,
	serverSelectionTimeoutMS: 5000
});

const userSchema = new mongoose.Schema(
	{
		email: { type: String, required: true, unique: true, lowercase: true, trim: true },
		name: { type: String, required: true, maxlength: 64 },
		points: { type: Number, default: 0, min: 0 },
		tags: [{ type: String }]
	},
	{ timestamps: true, collection: "users" }
);

const User = mongoose.model("User", userSchema);

// 创建（会自动执行 schema 校验）
const user = await User.create({ email: "a@example.com", name: "张三" });

// 查询：lean() 返回普通对象，性能更好
const list = await User.find({ points: { $gte: 10 } }).lean();

// 更新：new: true 返回更新后的文档
const updated = await User.findOneAndUpdate(
	{ email: "a@example.com" },
	{ $inc: { points: 1 } },
	{ new: true }
);

// 删除
await User.deleteOne({ _id: user._id });

// 关闭
await mongoose.disconnect();

与官方驱动的差异：

能力	官方驱动	Mongoose
Schema 校验	无（用 JSON Schema）	有，写入前校验
类型转换	手动	自动（字符串转数字等）
关联	`$lookup`	`populate()`（本质是二次查询）
性能开销	低	略高（文档包装、getter/setter）
逃生通道	—	`Model.collection` 拿原生集合

lean() 与 populate() 的注意点：

lean() 后没有实例方法，也不能 save()。
populate() 是 N+1 查询，列表页慎用；需要一次关联时改用聚合 $lookup。

十二、连接池与优雅关闭

process.on("SIGTERM", async () => {
	try {
		await client.close(); // 关闭所有连接
		process.exit(0);
	} catch (error) {
		console.error("关闭失败", error);
		process.exit(1);
	}
});

健康检查：

async function checkMongo() {
	try {
		await client.db("admin").command({ ping: 1 });
		return true;
	} catch {
		return false;
	}
}

ping 走的是已有连接，比 listCollections 轻。

十三、常见问题

现象	原因	处理
启动后第一个请求很慢	连接惰性建立	启动时显式 `connect()`
`Cannot use a session that has ended`	事务结束后仍使用 session	在 `withTransaction` 回调内使用
查询返回 `null`	类型不匹配（字符串 vs ObjectId）	先 `new ObjectId()` 并校验
`MongoServerSelectionError`	DNS/网络/副本集配置	检查连接串、`replicaSet` 参数
内存持续上涨	`toArray()` 加载大结果集	改用游标
写入成功但查不到	写到了不同库/集合	核对 `db()` 与集合名
更新时间差 8 小时	展示层未转换时区	存 UTC，展示时格式化
唯一索引不生效	已有重复数据导致索引创建失败	先清理重复再建索引

十四、FAQ

Q1：一个进程需要几个 MongoClient？

一个。MongoClient 自带连接池，每个请求新建会导致连接爆炸。

Q2：findOneAndUpdate 为什么拿不到 { value }？

6.x 默认直接返回文档。需要元数据时加 includeResultMetadata: true。

Q3：skip 分页慢怎么办？

改用基于排序字段的游标分页（createdAt < last），避免扫描并丢弃前 N 条。

Q4：事务能不能跨数据库？

同一副本集内的多个库可以；跨分片事务性能代价大，尽量避免。

Q5：什么时候用 bulkWrite？

批量导入或批量更新时，减少往返次数：

await db.collection("users").bulkWrite([
	{ insertOne: { document: { email: "a@example.com" } } },
	{ updateOne: { filter: { email: "b@example.com" }, update: { $set: { status: "active" } } } }
]);

Q6：ObjectId 能直接当字符串比较吗？

不能。ObjectId 与字符串在 BSON 中是不同类型，查询不会匹配。

Q7：需要强一致读怎么办？

事务内用 { readConcern: { level: "majority" } }，或读主节点（readPreference: "primary"）。

Q8：Mongoose 和原生驱动能混用吗？

可以，Model.collection 暴露原生集合。但同一集合混用两套写入路径会让校验与类型转换失效，需要明确约定。

Q9：如何做软删除？

加 deletedAt 字段，查询统一带 { deletedAt: null }，并建部分索引 partialFilterExpression: { deletedAt: null }。

Q10：文档超过 16MB 怎么办？

拆分文档（把数组拆成独立集合）或用 GridFS 存大文件。

附录：API 速查

const { MongoClient, ObjectId } = require("mongodb");

// 连接
const client = new MongoClient(uri, { maxPoolSize: 10, serverSelectionTimeoutMS: 5000 });
await client.connect();
const db = client.db("appdb");
await client.close();

// 集合
const col = db.collection("users");
await db.createCollection("users", { validator: {} });
await col.drop();

// 增
await col.insertOne({ ... });
await col.insertMany([{ ... }], { ordered: false });
await col.bulkWrite([{ insertOne: { document: {} } }]);

// 查
await col.findOne({ _id: new ObjectId(id) });
await col.find({ status: "active" }, { projection: { name: 1 } }).sort({ createdAt: -1 }).limit(20).toArray();
for await (const doc of col.find({}).batchSize(500)) { }
await col.countDocuments({});

// 改
await col.updateOne({ _id }, { $set: { name: "x" } });
await col.updateMany({ status: "x" }, { $inc: { n: 1 } });
await col.findOneAndUpdate({ _id }, { $set: {} }, { returnDocument: "after" });
await col.replaceOne({ _id }, { name: "x" });

// 删
await col.deleteOne({ _id });
await col.deleteMany({ status: "x" });

// 聚合
await col.aggregate([{ $match: {} }, { $group: { _id: "$f", n: { $sum: 1 } } }]).toArray();

// 索引
await col.createIndex({ email: 1 }, { unique: true });
await col.indexes();
await col.dropIndex("email_1");

// 事务
const session = client.startSession();
await session.withTransaction(async () => { /* 带 { session } 的操作 */ });
await session.endSession();

// 变更流
const stream = col.watch();
stream.on("change", change => {});
await stream.close();