這一篇是“100天精通Python”系列的第 40 天主題非常直接用pymongo操作 MongoDB 數(shù)據(jù)庫(kù)從安裝、連接到增刪改查再到批量任務(wù)和索引優(yōu)化最后帶上接口封裝思路和常見(jiàn)問(wèn)題排查。MongoDB 是目前最常用的文檔型 NoSQL 數(shù)據(jù)庫(kù)數(shù)據(jù)以 BSON 格式存儲(chǔ)業(yè)務(wù)上可以理解成“存在數(shù)據(jù)庫(kù)里的 JSON 對(duì)象”。Python 操作 MongoDB 最主流的方案就是官方驅(qū)動(dòng)pymongo它的 API 設(shè)計(jì)比較簡(jiǎn)潔和 MongoDB 的 Shell 語(yǔ)法很接近適合做爬蟲(chóng)存儲(chǔ)、日志收集、用戶(hù)行為分析、配置中心這類(lèi)場(chǎng)景。本文不繞彎子直接分成幾個(gè)部分先看 pymongo 的核心能力、再準(zhǔn)備環(huán)境、然后啟動(dòng) MongoDB 服務(wù)、接著用代碼把增刪改查和批量任務(wù)全部跑一遍最后給出索引優(yōu)化、接口 API 封裝和常見(jiàn)坑的排查方法。如果你正在學(xué) Python 操作數(shù)據(jù)庫(kù)或者準(zhǔn)備把項(xiàng)目里的數(shù)據(jù)存儲(chǔ)切到 MongoDB這篇文章可以直接作為操作手冊(cè)。1. 核心能力速覽能力項(xiàng)說(shuō)明項(xiàng)目類(lèi)型Python 官方 MongoDB 驅(qū)動(dòng)程序核心功能連接 MongoDB、數(shù)據(jù)庫(kù)/集合管理、增刪改查、聚合管道、索引管理、批量寫(xiě)入、事務(wù)支持支持平臺(tái)Windows、Linux、macOSPython 版本建議 Python 3.8 及以上具體以 pymongo 當(dāng)前版本要求為準(zhǔn)MongoDB 版本支持 MongoDB 3.6 及以上新版 pymongo 建議配合 MongoDB 4.2 使用安裝方式pip install pymongo通信方式TCP默認(rèn)端口 27017批量能力insert_many、bulk_write、游標(biāo)批量遍歷接口 API配合 Flask/FastAPI 可封裝成 HTTP 接口適合場(chǎng)景爬蟲(chóng)數(shù)據(jù)存儲(chǔ)、日志收集、用戶(hù)行為記錄、配置中心、快速原型開(kāi)發(fā)不適合場(chǎng)景復(fù)雜多表關(guān)聯(lián)查詢(xún)、強(qiáng)事務(wù)一致性、圖關(guān)系分析從材料看pymongo 的定位是“用 Python 的方式操作 MongoDB”所以你在 MongoDB Shell 里能做的操作在 pymongo 里基本都能對(duì)應(yīng)著寫(xiě)出來(lái)。安裝成本很低沒(méi)有額外的編譯依賴(lài)只要有網(wǎng)絡(luò)環(huán)境一條命令就能裝好。2. 適用場(chǎng)景與使用邊界先判斷這個(gè)工具適不適合你。MongoDB 的文檔模型比關(guān)系型表結(jié)構(gòu)更靈活一個(gè)集合collection里的文檔不要求字段完全一致新增字段不需要先改表結(jié)構(gòu)。這種特性非常適合數(shù)據(jù)格式多變、字段可能長(zhǎng)期演進(jìn)的業(yè)務(wù)比如爬蟲(chóng)保存的頁(yè)面數(shù)據(jù)、埋點(diǎn)日志、用戶(hù)配置信息。pymongo 在這種場(chǎng)景下能把 Python 字典直接寫(xiě)入數(shù)據(jù)庫(kù)讀出來(lái)也是字典和 Python 對(duì)象之間的轉(zhuǎn)換成本很低。但也要注意邊界。如果你需要大量JOIN操作、多行事務(wù)、復(fù)雜聚合統(tǒng)計(jì)MongoDB 雖然支持一部分但并不是最優(yōu)選擇。傳統(tǒng) ERP、金融賬務(wù)系統(tǒng)還是建議用 MySQL 或 PostgreSQL。pymongo 只是數(shù)據(jù)庫(kù)驅(qū)動(dòng)它本身不會(huì)幫你解決數(shù)據(jù)建模問(wèn)題集合設(shè)計(jì)不好同樣的查詢(xún)會(huì)明顯變慢。使用邊界方面重點(diǎn)提三點(diǎn)不要把 MongoDB 服務(wù)直接暴露到公網(wǎng)必須開(kāi)啟訪(fǎng)問(wèn)認(rèn)證否則很容易被掃描和寫(xiě)入惡意數(shù)據(jù)。涉及用戶(hù)隱私、電話(huà)號(hào)碼、身份信息等敏感字段建議在寫(xiě)入前加密或脫敏至少不要明文保存在沒(méi)有權(quán)限控制的集合里。要定期備份數(shù)據(jù)特別是使用副本集或單機(jī)部署時(shí)否則誤操作delete_many后恢復(fù)成本很高。3. 環(huán)境準(zhǔn)備與前置條件在寫(xiě)代碼之前需要把 Python、MongoDB、pymongo 三樣?xùn)|西準(zhǔn)備好。3.1 安裝 MongoDBMongoDB 需要單獨(dú)安裝。Windows 用戶(hù)可以下載 MongoDB Community Server 的 MSI 安裝包安裝時(shí)選擇“Install MongoDB as a Service”可以注冊(cè)成 Windows 服務(wù)開(kāi)機(jī)自啟。Linux 用戶(hù)可以通過(guò) apt 或 yum 安裝macOS 用戶(hù)可以用 Homebrew。安裝完成后確認(rèn)mongod命令可用。Windows 下如果配置了服務(wù)可以直接啟動(dòng)服務(wù)如果沒(méi)配置可以手動(dòng)指定數(shù)據(jù)目錄啟動(dòng)# Linux/macOS 示例需要先創(chuàng)建數(shù)據(jù)目錄 mkdir -p /data/db mongod --dbpath /data/db --port 27017Windows 命令行啟動(dòng)方式mongod --dbpath D:\mongodb\data --port 27017啟動(dòng)日志中出現(xiàn)Waiting for connections說(shuō)明服務(wù)已經(jīng)正常監(jiān)聽(tīng) 27017 端口。如果只想快速驗(yàn)證也可以不用本地安裝直接在測(cè)試服務(wù)器上用 Docker 啟動(dòng)docker run -d --name mongodb -p 27017:27017 mongo:6.0mongo:6.0是示例鏡像標(biāo)簽具體版本可以根據(jù)自己的環(huán)境調(diào)整。3.2 安裝 Python 和 pymongoPython 建議用 3.8 以上版本。先確認(rèn) Python 環(huán)境python --version然后安裝 pymongopip install pymongo如果需要操作 MongoDB 的聚合管道、GridFS、加密等擴(kuò)展功能還可以安裝pymongo[srv]或pymongo[encryption]但基礎(chǔ)使用只需要一個(gè)pymongo包。驗(yàn)證安裝是否成功python -c import pymongo; print(pymongo.version)能輸出版本號(hào)說(shuō)明驅(qū)動(dòng)已經(jīng)可用。3.3 檢查 MongoDB 服務(wù)狀態(tài)在 Python 中連接之前建議先用命令行工具mongosh或者直接通過(guò) pymongo 檢查。最簡(jiǎn)單的檢查方式是寫(xiě)一段連接代碼后面會(huì)講到。只要端口沒(méi)被防火墻攔截連接本身不會(huì)太復(fù)雜。4. 連接 MongoDB 數(shù)據(jù)庫(kù)連接 MongoDB 的核心類(lèi)是MongoClient。這一步要解決兩件事如何連接本地服務(wù)、如何帶認(rèn)證信息連接。4.1 創(chuàng)建 MongoClient 連接先把代碼寫(xiě)出來(lái)from pymongo import MongoClient # 本地默認(rèn)端口連接 client MongoClient(mongodb://127.0.0.1:27017/) # 獲取數(shù)據(jù)庫(kù)不存在會(huì)自動(dòng)創(chuàng)建 db client.mydb # 獲取集合不存在會(huì)自動(dòng)創(chuàng)建 collection db.users這里client.mydb和client[mydb]是等價(jià)的。MongoDB 是“懶創(chuàng)建”機(jī)制只有實(shí)際執(zhí)行寫(xiě)入操作時(shí)數(shù)據(jù)庫(kù)和集合才會(huì)真正創(chuàng)建。4.2 帶用戶(hù)名密碼的連接如果 MongoDB 開(kāi)啟了認(rèn)證就需要在連接串里帶上賬號(hào)client MongoClient( mongodb://admin:password123127.0.0.1:27017/?authSourceadmin )authSource表示認(rèn)證庫(kù)常見(jiàn)為admin。如果你的賬號(hào)建立在某個(gè)業(yè)務(wù)庫(kù)下authSource就寫(xiě)對(duì)應(yīng)的庫(kù)名。生產(chǎn)環(huán)境建議用環(huán)境變量保存連接串不要硬編碼到代碼里。4.3 連接參數(shù)MongoClient還支持一些常用參數(shù)client MongoClient( mongodb://127.0.0.1:27017/, serverSelectionTimeoutMS5000, # 5秒內(nèi)選不到服務(wù)器就報(bào)錯(cuò) connectTimeoutMS3000, maxPoolSize50 )maxPoolSize控制連接池大小。默認(rèn)的MongoClient是懶連接真正發(fā)起操作時(shí)才會(huì)建立連接。如果你在腳本里只是創(chuàng)建 client 并不操作它不會(huì)立刻報(bào)錯(cuò)。4.4 驗(yàn)證連接是否成功from pymongo import MongoClient client MongoClient(mongodb://127.0.0.1:27017/, serverSelectionTimeoutMS5000) # 向服務(wù)器發(fā)送 ping 命令 try: client.admin.command(ping) print(MongoDB 連接成功) except Exception as e: print(連接失敗:, e)這個(gè)ping命令是驗(yàn)證網(wǎng)絡(luò)和服務(wù)狀態(tài)最直接的方式。如果失敗先檢查服務(wù)有沒(méi)有啟動(dòng)、端口是否正確。5. 文檔插入操作MongoDB 的基本數(shù)據(jù)單元是文檔在 pymongo 中對(duì)應(yīng) Python 字典。插入操作有insert_one和insert_many。5.1 插入單條文檔from pymongo import MongoClient from datetime import datetime client MongoClient(mongodb://127.0.0.1:27017/) db client.mydb collection db.users user { name: 張三, age: 28, email: zhangsanexample.com, tags: [python, mongodb], created_at: datetime.now() } result collection.insert_one(user) print(result.inserted_id)插入成功后result.inserted_id會(huì)返回自動(dòng)生成的ObjectId。如果你在文檔里手動(dòng)指定了_id字段MongoDB 會(huì)使用你指定的值。5.2 批量插入文檔users [ {name: 李四, age: 32, city: 上海}, {name: 王五, age: 24, city: 北京}, {name: 趙六, age: 29, city: 廣州} ] result collection.insert_many(users) print(result.inserted_ids)insert_many接受一個(gè)列表返回一個(gè)包含所有_id的列表。批量插入比循環(huán)insert_one效率高很多因?yàn)闇p少了客戶(hù)端和服務(wù)端的往返次數(shù)。5.3 插入數(shù)據(jù)后的驗(yàn)證count collection.count_documents({}) print(當(dāng)前用戶(hù)數(shù)量:, count)注意count()方法在新版 pymongo 中已經(jīng)棄用統(tǒng)一使用count_documents。6. 文檔查詢(xún)操作查詢(xún)是 pymongo 里用得最多的部分。掌握查詢(xún)語(yǔ)法基本就能解決 80% 的日常需求。6.1 查詢(xún)單條文檔find_oneuser collection.find_one({name: 張三}) print(user)返回的是字典找不到時(shí)返回None。和 MongoDB Shell 的findOne行為一致。6.2 查詢(xún)多條文檔findcursor collection.find({age: {$gte: 25}}) for user in cursor: print(user[name], user[age])find返回一個(gè)Cursor對(duì)象是一個(gè)可迭代的游標(biāo)?？梢杂胠ist()轉(zhuǎn)換成列表但如果數(shù)據(jù)量很大不建議一次性全部加載到內(nèi)存。6.3 常用比較條件操作符含義示例$eq等于{age: {$eq: 28}}$ne不等于{age: {$ne: 28}}$gt大于{age: {$gt: 25}}$gte大于等于{age: {$gte: 25}}$lt小于{age: {$lt: 30}}$lte小于等于{age: {$lte: 30}}$in在列表中{city: {$in: [北京, 上海]}}$nin不在列表中{city: {$nin: [北京]}}$regex正則匹配{name: {$regex: ^張}}# 查詢(xún)年齡在 25 到 30 之間或者城市為上海的用戶(hù) query { $or: [ {age: {$gte: 25, $lte: 30}}, {city: 上海} ] } for user in collection.find(query): print(user)6.4 排序、分頁(yè)、計(jì)數(shù)# 按年齡降序排序 cursor collection.find().sort(age, -1) # 跳過(guò)前2條只取3條 cursor collection.find().sort(age, -1).skip(2).limit(3) # 只返回 name 和 age 字段不返回 _id cursor collection.find({}, {_id: 0, name: 1, age: 1})skip加limit是傳統(tǒng)分頁(yè)方案。數(shù)據(jù)量超過(guò)幾萬(wàn)頁(yè)時(shí)更推薦用_id或排序字段做游標(biāo)分頁(yè)避免大偏移量導(dǎo)致性能下降。6.5 正則和數(shù)組查詢(xún)# 名字以 張 開(kāi)頭 users collection.find({name: {$regex: ^張}}) # tags 數(shù)組中包含 python users collection.find({tags: python}) # tags 數(shù)組同時(shí)包含 python 和 mongodb users collection.find({tags: {$all: [python, mongodb]}})數(shù)組查詢(xún)?cè)跇?biāo)簽系統(tǒng)、權(quán)限系統(tǒng)里很常用$all表示數(shù)組字段必須同時(shí)包含多個(gè)值。7. 文檔更新操作更新操作最常用的是update_one和update_many。更新時(shí)要注意直接傳入整個(gè)新文檔會(huì)替換原文檔更新指定字段必須使用更新操作符比如$set、$inc、$unset、$push、$addToSet。7.1 修改指定字段update_oneresult collection.update_one( {name: 張三}, {$set: {age: 29, city: 深圳}} ) print(result.matched_count, result.modified_count)matched_count表示匹配到的文檔數(shù)modified_count表示實(shí)際被修改的文檔數(shù)。這里能發(fā)現(xiàn)如果新值和舊值一樣modified_count是 0。7.2 批量更新update_many# 將所有北京用戶(hù)的年齡加 1 result collection.update_many( {city: 北京}, {$inc: {age: 1}} ) print(result.modified_count)$inc是數(shù)值增減操作適合計(jì)數(shù)、積分、庫(kù)存等場(chǎng)景。7.3 數(shù)組更新$push和$addToSet# 向 tags 數(shù)組追加一個(gè)值 collection.update_one( {name: 張三}, {$push: {tags: pymongo}} ) # 如果值已經(jīng)存在則不重復(fù)添加 collection.update_one( {name: 張三}, {$addToSet: {tags: pymongo}} )$push每次都會(huì)追加$addToSet會(huì)去重。需要維護(hù)標(biāo)簽、關(guān)注列表、操作日志時(shí)優(yōu)先考慮$addToSet。7.4 替換整條文檔replace_onenew_doc { name: 張三, age: 30, city: 廣州, tags: [python] } collection.replace_one({name: 張三}, new_doc)replace_one會(huì)刪除原文檔中其他字段用新文檔整體替換。除非業(yè)務(wù)明確需要否則更推薦$set做局部更新避免誤刪字段。7.5 upsert 更新或插入result collection.update_one( {name: 孫七}, {$set: {age: 18, city: 杭州}}, upsertTrue ) print(result.upserted_id)加上upsertTrue后如果查詢(xún)條件匹配不到文檔就會(huì)插入一條新文檔。這個(gè)能力在“存在則更新不存在則創(chuàng)建”的業(yè)務(wù)中非常實(shí)用。8. 文檔刪除操作刪除操作分為delete_one和delete_many。# 刪除名字為張三的第一條記錄 result collection.delete_one({name: 張三}) print(result.deleted_count) # 刪除所有城市為北京的用戶(hù) result collection.delete_many({city: 北京}) print(result.deleted_count) # 清空集合保留集合本身 collection.delete_many({})刪除整個(gè)集合可以用collection.drop()清除整個(gè)數(shù)據(jù)庫(kù)client.drop_database(mydb)刪除是不可逆操作建議在執(zhí)行delete_many前先用count_documents看一下確定影響范圍或者在測(cè)試庫(kù)上先驗(yàn)證。9. 索引與批量任務(wù)索引是 MongoDB 查詢(xún)性能的核心。沒(méi)有索引時(shí)MongoDB 必須做全表掃描數(shù)據(jù)到幾十萬(wàn)條后查詢(xún)延遲會(huì)明顯上升。9.1 創(chuàng)建索引# 單字段索引 collection.create_index(name) # 復(fù)合索引 collection.create_index([(city, 1), (age, -1)]) # 唯一索引 collection.create_index(email, uniqueTrue) # 查看集合索引 for index in collection.list_indexes(): print(index)1表示升序-1表示降序。如果業(yè)務(wù)查詢(xún)經(jīng)常同時(shí)按city和age過(guò)濾排序適合建復(fù)合索引。唯一索引可以用來(lái)防止郵箱、手機(jī)號(hào)等字段重復(fù)。9.2 批量寫(xiě)入與bulk_write批量任務(wù)指的是在一次請(qǐng)求里執(zhí)行多條寫(xiě)操作。前面提到的insert_many只能批量插入而bulk_write可以混合插入、更新、刪除。from pymongo import InsertOne, UpdateOne, DeleteOne operations [ InsertOne({name: 測(cè)試1, age: 20}), UpdateOne({name: 測(cè)試2}, {$set: {age: 21}}, upsertTrue), DeleteOne({name: 測(cè)試3}) ] result collection.bulk_write(operations) print(result.inserted_count) print(result.modified_count) print(result.deleted_count)bulk_write適合做數(shù)據(jù)同步、清洗、ETL 任務(wù)。它默認(rèn)按順序執(zhí)行如果中間某條失敗后面的操作可能不會(huì)繼續(xù)可以使用orderedFalse讓 MongoDB 繼續(xù)執(zhí)行剩余操作result collection.bulk_write(operations, orderedFalse)9.3 批量查詢(xún)與游標(biāo)遍歷如果要對(duì)全量數(shù)據(jù)做處理不要一次性list()出來(lái)應(yīng)該使用游標(biāo)分塊遍歷batch_size 500 cursor collection.find({}).batch_size(batch_size) for doc in cursor: # 處理每條文檔 process(doc)batch_size控制每次從服務(wù)端取回的文檔數(shù)量可以避免把大量數(shù)據(jù)一次性加載到內(nèi)存。10. 接口 API 調(diào)用示例pymongo 本身不是 HTTP 服務(wù)但實(shí)際項(xiàng)目中經(jīng)常需要把數(shù)據(jù)庫(kù)操作封裝成接口讓前端或內(nèi)部服務(wù)調(diào)用。下面用 Flask 做一個(gè)最簡(jiǎn)單的示例演示如何把 pymongo 繼承到 API 層。pip install flask pymongofrom flask import Flask, request, jsonify from pymongo import MongoClient from bson import ObjectId app Flask(__name__) client MongoClient(mongodb://127.0.0.1:27017/) db client.mydb collection db.users def serialize_doc(doc): 把 ObjectId 轉(zhuǎn)成字符串方便 JSON 返回 if doc is None: return None doc[_id] str(doc[_id]) return doc app.route(/users, methods[GET]) def list_users(): users list(collection.find({})) return jsonify([serialize_doc(user) for user in users]) app.route(/users, methods[POST]) def create_user(): data request.get_json() if not data or not data.get(name): return jsonify({error: name is required}), 400 result collection.insert_one(data) return jsonify({_id: str(result.inserted_id)}), 201 app.route(/users/user_id, methods[PUT]) def update_user(user_id): data request.get_json() result collection.update_one( {_id: ObjectId(user_id)}, {$set: data} ) if result.matched_count 0: return jsonify({error: user not found}), 404 return jsonify({message: updated}) app.route(/users/user_id, methods[DELETE]) def delete_user(user_id): result collection.delete_one({_id: ObjectId(user_id)}) if result.deleted_count 0: return jsonify({error: user not found}), 404 return jsonify({message: deleted}) if __name__ __main__: app.run(host127.0.0.1, port5000, debugTrue)這是一個(gè)通用示例實(shí)際部署時(shí)要注意接口必須做權(quán)限校驗(yàn)不能把數(shù)據(jù)庫(kù)接口直接裸露給公網(wǎng)。入?yún)⒁ｒ?yàn)避免用戶(hù)傳入$set之類(lèi)的內(nèi)容修改非預(yù)期字段。ObjectId轉(zhuǎn)換失敗時(shí)要做異常處理。大規(guī)模接口服務(wù)建議使用 FastAPI Pydantic做自動(dòng)化參數(shù)校驗(yàn)和文檔輸出。也可以用curl快速驗(yàn)證接口curl http://127.0.0.1:5000/userscurl -X POST http://127.0.0.1:5000/users \ -H Content-Type: application/json \ -d {name: 接口用戶(hù), age: 18}11. 資源占用與性能觀察MongoDB 是內(nèi)存友好的數(shù)據(jù)庫(kù)嗎實(shí)際上它依賴(lài)操作系統(tǒng)緩存索引和熱數(shù)據(jù)盡可能放在內(nèi)存中查詢(xún)速度很快。單機(jī)部署時(shí)MongoDB 進(jìn)程會(huì)占用較多內(nèi)存這是正?，F(xiàn)象因?yàn)?LRU 緩存會(huì)主動(dòng)利用空閑內(nèi)存。觀察資源占用時(shí)重點(diǎn)看以下指標(biāo)觀察項(xiàng)方法說(shuō)明MongoDB 內(nèi)存占用任務(wù)管理器或mongostat緩存大小會(huì)動(dòng)態(tài)變化連接數(shù)db.serverStatus().connections連接池如果耗盡請(qǐng)求會(huì)阻塞查詢(xún)耗時(shí)explain(executionStats)檢查是否走索引磁盤(pán) IOiostat或云監(jiān)控寫(xiě)入量大時(shí)磁盤(pán)容易成為瓶頸Python 端連接池maxPoolSize高并發(fā)場(chǎng)景要合理設(shè)置explain是排查慢查詢(xún)的重要工具cursor collection.find({city: 北京}).sort(age, -1) explain_result cursor.explain() print(explain_result[queryPlanner][winningPlan])如果 winning plan 顯示COLLSCAN說(shuō)明沒(méi)有走索引。這時(shí)應(yīng)該考慮建立合適的索引。使用 pymongo 時(shí)有幾個(gè)性能習(xí)慣一個(gè)進(jìn)程內(nèi)復(fù)用同一個(gè)MongoClient不要每次操作都新建連接。批量插入優(yōu)先insert_many不要循環(huán)insert_one。查詢(xún)時(shí)只返回需要的字段減少網(wǎng)絡(luò)傳輸。避免一次find后把所有結(jié)果轉(zhuǎn)成 list改用游標(biāo)。寫(xiě)入量大的業(yè)務(wù)開(kāi)啟確認(rèn)寫(xiě)w1即可不要用wmajority做不必要的確認(rèn)除非你確實(shí)需要強(qiáng)一致。12. 常見(jiàn)問(wèn)題與排查方法問(wèn)題現(xiàn)象可能原因排查方式解決方案連接超時(shí)ServerSelectionTimeoutErrorMongoDB 服務(wù)未啟動(dòng)、端口錯(cuò)誤、防火墻攔截檢查mongod進(jìn)程、netstat -ano | findstr 27017、Telnet 端口啟動(dòng)服務(wù)確認(rèn)連接串端口放行防火墻認(rèn)證失敗Authentication failed用戶(hù)名密碼錯(cuò)誤、認(rèn)證庫(kù)不對(duì)、用戶(hù)沒(méi)有權(quán)限用mongosh測(cè)試認(rèn)證信息確認(rèn)authSource指向正確的認(rèn)證庫(kù)重新設(shè)置用戶(hù)插入文檔報(bào)DocumentTooLarge單條文檔超過(guò) 16 MB 限制查看報(bào)錯(cuò)內(nèi)容壓縮字段、拆分文檔或使用 GridFS 存儲(chǔ)大文件count()報(bào)錯(cuò)或告警新版 pymongo 移除了舊count查看版本變更改用count_documents({})查詢(xún)速度很慢沒(méi)有索引或索引不合理執(zhí)行explain查看COLLSCAN創(chuàng)建單字段或復(fù)合索引批量寫(xiě)入部分失敗orderedTrue模式下中途出錯(cuò)檢查報(bào)錯(cuò)開(kāi)啟日志使用orderedFalse或?qū)﹀e(cuò)誤記錄重試連接數(shù)過(guò)多每次操作創(chuàng)建新MongoClient且未復(fù)用查看連接數(shù)復(fù)用全局 client調(diào)整maxPoolSize數(shù)據(jù)亂碼字符集問(wèn)題檢查寫(xiě)入編碼客戶(hù)端統(tǒng)一使用 UTF-8刪除后數(shù)據(jù)無(wú)法恢復(fù)沒(méi)有備份檢查備份策略使用mongodump定期備份部署副本集最常見(jiàn)的問(wèn)題是“明明本地裝了 MongoDB但 Python 連不上”絕大多數(shù)原因是 MongoDB 服務(wù)沒(méi)有啟動(dòng)或者連接串端口寫(xiě)錯(cuò)。建議先跑一下ping命令再排查其他環(huán)節(jié)。13. 最佳實(shí)踐與使用建議pymongo 的 API 雖然簡(jiǎn)單但項(xiàng)目落地時(shí)還是有一些工程化經(jīng)驗(yàn)可以提前規(guī)避問(wèn)題。第一連接管理要統(tǒng)一。建議在項(xiàng)目入口創(chuàng)建全局MongoClient通過(guò)配置模塊讀取連接串不要在每個(gè)函數(shù)里都MongoClient(...)。這樣既能復(fù)用連接池也方便切測(cè)試庫(kù)和生產(chǎn)庫(kù)。# config.py import os from pymongo import MongoClient MONGO_URI os.getenv(MONGO_URI, mongodb://127.0.0.1:27017/) client MongoClient(MONGO_URI) db client.mydb第二集合名和字段名要有規(guī)范。集合名建議使用復(fù)數(shù)名詞比如users、orders。字段名統(tǒng)一使用小寫(xiě)和下劃線(xiàn)避免出現(xiàn)userId和user_id混用。字段名風(fēng)格不統(tǒng)一后期寫(xiě)查詢(xún)條件時(shí)會(huì)非常痛苦。第三建立索引要有計(jì)劃。不要上來(lái)就給所有字段加索引索引會(huì)占用磁盤(pán)空間并且降低寫(xiě)入性能。應(yīng)該先分析查詢(xún)條件對(duì)高頻過(guò)濾字段建立索引對(duì)排序字段建立復(fù)合索引。第四生產(chǎn)環(huán)境一定要開(kāi)啟認(rèn)證和訪(fǎng)問(wèn)控制。MongoDB 默認(rèn)是沒(méi)有賬號(hào)密碼的如果監(jiān)聽(tīng)在0.0.0.0相當(dāng)于把數(shù)據(jù)裸奔在公網(wǎng)上。至少要做兩件事服務(wù)監(jiān)聽(tīng)內(nèi)網(wǎng) IP創(chuàng)建專(zhuān)用業(yè)務(wù)賬號(hào)并分配最小權(quán)限。第五涉及敏感數(shù)據(jù)時(shí)要加密和脫敏。不能把email、phone、id_card這類(lèi)字段明文存儲(chǔ)除非數(shù)據(jù)庫(kù)網(wǎng)絡(luò)和權(quán)限完全隔離。建議寫(xiě)入前用加密算法處理展示時(shí)再解密或者完成業(yè)務(wù)后自動(dòng)清理。第六批量任務(wù)要加上日志和失敗重試機(jī)制。用bulk_write處理大量數(shù)據(jù)時(shí)如果中間出錯(cuò)最好捕獲異常并記錄失敗的文檔 key后續(xù)單獨(dú)重試。不要在一次任務(wù)里寫(xiě)入上萬(wàn)條還不打印任何日志出了問(wèn)題很難定位。第七備份策略要提前設(shè)計(jì)。至少使用mongodump做定時(shí)全量備份有條件的部署副本集實(shí)現(xiàn)高可用。對(duì)于核心業(yè)務(wù)還可以開(kāi)啟 op log用增量備份降低數(shù)據(jù)丟失風(fēng)險(xiǎn)。14. 總結(jié)與下一步這一篇把 Python 操作 MongoDB 的完整路徑過(guò)了一遍環(huán)境準(zhǔn)備、服務(wù)啟動(dòng)、pymongo 連接、插入、查詢(xún)、更新、刪除、索引、批量任務(wù)、接口封裝和排錯(cuò)方法。最值得先動(dòng)手驗(yàn)證的是insert_onefind_oneupdate_onedelete_one這條基礎(chǔ)鏈路跑通之后MongoDB 的核心用法你就已經(jīng)掌握了。最容易踩的坑有三個(gè)第一MongoDB 服務(wù)沒(méi)啟動(dòng)就急著跑 Python 代碼結(jié)果報(bào)連接超時(shí)第二更新文檔時(shí)忘記用$set結(jié)果整條文檔被覆蓋第三大量數(shù)據(jù)查詢(xún)時(shí)沒(méi)有建索引數(shù)據(jù)量上去之后接口變慢。如果這周有時(shí)間建議繼續(xù)往下練幾個(gè)方向MongoDB 聚合管道aggregate做分組統(tǒng)計(jì)GridFS存大文件副本集事務(wù)操作以及motor異步驅(qū)動(dòng)對(duì)接 FastAPI。等基礎(chǔ)增刪改查熟練之后再去看聚合和事務(wù)會(huì)輕松很多。建議把這篇文章收藏備用下次需要操作 MongoDB 時(shí)可以直接對(duì)照著寫(xiě)代碼。