冠军资讯
秃头程序员在实际的后端开发中，Python3 作为一种流行的脚本语言，经常需要与 MongoDB 这种 NoSQL 数据库进行交互。本文旨在解决 Python3 操作 MongoDB 过程中常见的连接、查询、更新等问题，并提供最佳实践方案。例如，在高并发场景下，如何优化连接池，防止出现连接耗尽的问题？本文将深入探讨这些问题，并提供可直接使用的代码示例。
MongoDB 连接配置与优化
首先，我们需要使用 pymongo 库来连接 MongoDB。确保已经安装了该库：
pip install pymongo
然后，可以使用以下代码建立连接：
from pymongo import MongoClient
# 连接 MongoDB，默认连接到 localhost:27017
client = MongoClient('mongodb://localhost:27017/') # 可以指定 host 和 port，例如：'mongodb://username:password@host:port/'
# 选择数据库
db = client['mydatabase']
# 选择集合（相当于关系型数据库中的表）
collection = db['mycollection']
# 打印服务器信息
print(client.server_info())
连接池优化：
在高并发场景下，频繁地创建和关闭连接会消耗大量的资源。因此，使用连接池是必不可少的。pymongo 默认已经实现了连接池，但我们可以通过调整一些参数来优化性能：
client = MongoClient('mongodb://localhost:27017/', maxPoolSize=20, connectTimeoutMS=30000, serverSelectionTimeoutMS=30000) # maxPoolSize 设置连接池大小，connectTimeoutMS 设置连接超时时间
maxPoolSize 建议根据服务器的硬件配置和应用的并发量进行调整。connectTimeoutMS 设置连接超时时间，防止程序长时间阻塞。
身份验证：
如果 MongoDB 启用了身份验证，需要在连接字符串中提供用户名和密码：
client = MongoClient('mongodb://username:password@localhost:27017/mydatabase')
MongoDB 常用操作：CRUD
创建（Create）：
# 插入单个文档
post = {"author": "Mike", "text": "My first blog post!", "tags": ["mongodb", "python", "pymongo"]}
post_id = collection.insert_one(post).inserted_id
print(f"Inserted document ID: {post_id}")
# 插入多个文档
new_posts = [
{"author": "Mike", "text": "Another post!", "tags": ["bulk", "insert"]},
{"author": "Eliot", "title": "MongoDB is fun", "text": "and pretty easy too!"}
]
result = collection.insert_many(new_posts)
print(f"Inserted document IDs: {result.inserted_ids}")
读取（Read）：
# 查询单个文档
print(collection.find_one({"author": "Mike"}))
# 查询所有文档
for post in collection.find():
print(post)
# 带条件查询
for post in collection.find({"author": "Mike", "tags": "mongodb"}):
print(post)
# 使用查询操作符（例如：$gt, $lt, $in）
for post in collection.find({"likes": {"$gt": 10}}): # 查询 likes 大于 10 的文档
print(post)
# 排序
for post in collection.find().sort("author", 1): # 按照 author 升序排序
print(post)
# 分页
for post in collection.find().skip(10).limit(5): # 跳过 10 条数据，取 5 条数据
print(post)
更新（Update）：
# 更新单个文档
result = collection.update_one({"author": "Mike"}, {"$set": {"text": "Updated text!"}})
print(f"Matched count: {result.matched_count}, Modified count: {result.modified_count}")
# 更新多个文档
result = collection.update_many({"author": "Mike"}, {"$set": {"text": "Updated text again!"}})
print(f"Matched count: {result.matched_count}, Modified count: {result.modified_count}")
# 使用 upsert，如果不存在则插入
result = collection.update_one({"author": "NonExistentAuthor"}, {"$set": {"text": "New document"}}, upsert=True)
print(f"Upserted ID: {result.upserted_id}")
删除（Delete）：
# 删除单个文档
result = collection.delete_one({"author": "Mike"})
print(f"Deleted count: {result.deleted_count}")
# 删除多个文档
result = collection.delete_many({"author": "Mike"})
print(f"Deleted count: {result.deleted_count}")
实战避坑经验
索引优化： 对于频繁查询的字段，建立索引可以显著提升查询效率。可以使用
create_index()方法创建索引：
collection.create_index([("author", 1)]) # 对 author 字段创建升序索引需要注意的是，过多的索引会降低写入性能，因此需要权衡。
批量操作： 对于大量数据的插入、更新和删除，使用
bulk_write()方法可以显著提升性能。这比循环调用insert_one()、update_one()和delete_one()效率更高。from pymongo import InsertOne, UpdateOne, DeleteOne requests = [ InsertOne({"author": "Mike", "text": "Bulk insert"}), UpdateOne({"author": "Mike"}, {"$set": {"text": "Bulk update"}}), DeleteOne({"author": "Mike"}) ] result = collection.bulk_write(requests) print(result.bulk_api_result)错误处理： 在实际应用中，需要对 MongoDB 操作进行错误处理，例如连接超时、网络异常等。可以使用
try...except语句捕获异常，并进行相应的处理。数据类型： 注意 Python 和 MongoDB 的数据类型差异。例如，Python 的
datetime对象需要转换为 MongoDB 的datetime类型。
使用 Nginx 反向代理： 如果你的 MongoDB 部署在云服务器上，并且需要对外提供服务，可以使用 Nginx 作为反向代理，实现负载均衡和安全防护。同时，需要注意配置防火墙规则，限制 MongoDB 的访问权限。合理设置 Nginx 的
worker_processes和worker_connections，可以提高并发连接数，在高流量场景下，也可以考虑使用宝塔面板简化 Nginx 的配置和管理。监控与日志： 监控 MongoDB 的性能指标（例如：CPU 使用率、内存使用率、磁盘 I/O）以及日志，可以帮助你及时发现和解决问题。可以使用 MongoDB 自带的监控工具，也可以使用第三方监控工具（例如：Prometheus + Grafana）。
通过本文，我们学习了如何使用 Python3 连接 MongoDB，并进行常见的 CRUD 操作。同时，我们也了解了一些实战避坑经验，可以帮助你在实际项目中更好地使用 MongoDB。
