引言
在 Python 中使用 PyMongo 進行 MongoDB 操作時，開發者需要注意執行緒與程序的差異。雖然 PyMongo 是執行緒安全的，但在多程序環境下卻不是程序安全的。本文將結合案例分析 PyMongo 為什麼不是程序安全的，同時說明其執行緒安全性。
PyMongo 的執行緒安全性
PyMongo 的 MongoClient 是執行緒安全的。多個執行緒可以共享同一個 MongoClient 例項，它透過內部的連線池管理機制確保執行緒安全。
示例：執行緒安全
以下程式碼展示瞭如何在多執行緒環境中安全使用 PyMongo：
from pymongo import MongoClient
import threading
# 建立一個全域性 MongoClient 例項
mongo_client = MongoClient("mongodb://localhost:27017")
collection = mongo_client["test_db"]["logs"]
def insert_log(log_entry):
"""插入日誌到 MongoDB"""
collection.insert_one({"log_entry": log_entry})
# 啟動多個執行緒
threads = []
for i in range(10):
thread = threading.Thread(target=insert_log, args=(f"Log entry {i}",))
threads.append(thread)
thread.start()
# 等待所有執行緒完成
for thread in threads:
thread.join()
mongo_client.close()
結果：
所有執行緒共享一個連線池。
MongoClient 透過內部鎖機制確保執行緒安全。
執行無錯誤。
PyMongo 執行緒安全的原因
MongoClient的內部連線池是執行緒安全的。多執行緒操作透過鎖機制和佇列管理連線分配，避免了競爭條件。
PyMongo 的程序不安全性
在多程序環境下，如果父程序建立了 MongoClient 並 fork 出多個子程序，子程序會繼承父程序的記憶體狀態，包括 MongoClient 和其連線池。然而，連線池中的 socket 是不可共享的，這就導致了以下問題：
連線衝突：子程序嘗試複用父程序的連線，可能導致資源競爭。
連線斷開：子程序無法正確管理這些連線，導致
AutoReconnect錯誤。連線數膨脹：每個子程序可能重新建立連線，導致連線數激增。
示例：程序不安全
以下程式碼展示了 PyMongo 在多程序環境中的問題：
from pymongo import MongoClient
from multiprocessing import Process
# 在父程序中建立 MongoClient
mongo_client = MongoClient("mongodb://localhost:27017")
collection = mongo_client["test_db"]["logs"]
def insert_log(log_entry):
"""插入日誌到 MongoDB"""
collection.insert_one({"log_entry": log_entry})
# 建立多個子程序
processes = []
for i in range(4):
process = Process(target=insert_log, args=(f"Log entry {i}",))
processes.append(process)
process.start()
# 等待所有子程序完成
for process in processes:
process.join()
mongo_client.close()結果：
可能丟擲以下警告或錯誤：
UserWarning: MongoClient opened before fork. pymongo.errors.AutoReconnect: Connection reset by peer.
每個子程序嘗試使用父程序的連線池，導致連線不安全。
PyMongo 為什麼不是程序安全的
連線池不共享
PyMongo 的
MongoClient例項中維護了一個連線池，用於管理與 MongoDB 的 socket 連線。當父程序 fork 子程序時，連線池的 socket 物件無法在子程序中正確複用，因為 socket 是作業系統級別的資源。
狀態衝突
子程序繼承了父程序的連線池狀態，但這些狀態對子程序而言是過期的或無效的，導致子程序無法正確管理連線。
競爭條件
子程序嘗試使用父程序的連線資源，可能導致資源競爭。
解決方案
方案 1：子程序中建立 MongoClient
在子程序中獨立建立 MongoClient，避免父子程序共享連線池。
from multiprocessing import Process
from pymongo import MongoClient
def insert_log(log_entry):
"""每個子程序獨立建立 MongoClient"""
mongo_client = MongoClient("mongodb://localhost:27017")
collection = mongo_client["test_db"]["logs"]
collection.insert_one({"log_entry": log_entry})
mongo_client.close()
# 建立多個子程序
processes = []
for i in range(4):
process = Process(target=insert_log, args=(f"Log entry {i}",))
processes.append(process)
process.start()
# 等待所有子程序完成
for process in processes:
process.join()
優點：
每個子程序有獨立的
MongoClient例項，無連線衝突。
缺點：
連線開銷較大。
方案 2：使用 Queue 解耦資料庫操作
透過 multiprocessing.Queue，將日誌解析與資料庫操作解耦。一個專門的程序負責 MongoDB 操作，其餘程序透過佇列傳遞資料。
from multiprocessing import Process, Queue
from pymongo import MongoClient
def db_worker(queue):
"""專門處理 MongoDB 寫入的程序"""
mongo_client = MongoClient("mongodb://localhost:27017")
collection = mongo_client["test_db"]["logs"]
while True:
log_entry = queue.get()
if log_entry is None: # 結束訊號
break
collection.insert_one({"log_entry": log_entry})
mongo_client.close()
def process_log(log_entry, queue):
"""子程序負責日誌解析，將結果放入佇列"""
queue.put(log_entry)
if __name__ == "__main__":
logs = [f"Log entry {i}" for i in range(10)]
queue = Queue()
# 啟動資料庫程序
db_process = Process(target=db_worker, args=(queue,))
db_process.start()
# 啟動日誌解析程序
processes = []
for log in logs:
process = Process(target=process_log, args=(log, queue))
processes.append(process)
process.start()
for process in processes:
process.join()
queue.put(None) # 傳送結束訊號
db_process.join()
優點：
單一資料庫程序管理 MongoDB 連線，避免衝突。
資料庫寫入邏輯集中，易於維護。
缺點：
通訊開銷增加。
總結
執行緒安全：
PyMongo 的
MongoClient是執行緒安全的，可在多執行緒中安全複用。程序不安全：
由於
fork的原因，MongoClient的連線池在多程序環境中無法安全複用。需要在子程序中重新建立
MongoClient或透過Queue解耦資料庫操作。推薦方案：
小規模併發：子程序中建立
MongoClient。高併發場景：使用
Queue解耦，集中管理資料庫連線。
透過理解 PyMongo 的特性和問題，我們可以在多執行緒和多程序環境中更高效地使用 MongoDB。