from pymongo import MongoClient
from pymongo.errors import WriteError
class RetriableWrite:
def __init__(self, client, db_name, collection_name):
self.client = client
self.db = client[db_name]
self.collection = self.db[collection_name]
def insert_retry(self, document):
"""尝试插入文档，如果失败则重试。"""
while True:
try:
self.collection.insert_one(document)
print("文档插入成功。")
break
except WriteError:
print("文档插入失败，将在5秒后重试。")
time.sleep(5)
def update_retry(self, query, update):
"""尝试更新文档，如果失败则重试。"""
while True:
try:
self.collection.update_one(query, update)
print("文档更新成功。")
break
except WriteError:
print("文档更新失败，将在5秒后重试。")
time.sleep(5)
def delete_retry(self, query):
"""尝试删除文档，如果失败则重试。"""
while True:
try:
self.collection.delete_one(query)
print("文档删除成功。")
break
except WriteError:
print("文档删除失败，将在5秒后重试。")
time.sleep(5)
# 使用示例
client = MongoClient('mongodb://localhost:27017/')
db_name = 'test_db'
collection_name = 'test_collection'
retry_handler = RetriableWrite(client, db_name, collection_name)
# 插入文档
document = {'name': 'Alice', 'age': 25}
retry_handler.insert_retry(document)
# 更新文档
query = {'name': 'Alice'}
update = {'$set': {'age': 26}}
retry_handler.update_retry(query, update)
# 删除文档
query = {'name': 'Alice'}
retry_handler.delete_retry(query)这个代码示例展示了如何在Python中使用pymongo库来实现MongoDB的可重试写入操作。它定义了一个RetriableWrite类，其中包含了插入、更新和删除操作的重试方法。如果写入操作因为某些原因失败，它会在5秒后重试。这是一个简单的错误处理模式，可以用于任何需要重试逻辑的场景。