关于使用python对mongo多线程更新数据

from pymongo import MongoClient
import threading
 
# MongoDB 配置
mongo_uri = 'mongodb://localhost:27017/'
mongo_db_name = 'my_db'
mongo_collection_name = 'my_coll'
 
# 连接 MongoDB
mongo_client = MongoClient(mongo_uri)
mongo_db = mongo_client[mongo_db_name]
mongo_coll = mongo_db[mongo_collection_name]
 
# 查询 MongoDB
mongo_query = {}
mongo_batch_size = 1000
mongo_results = mongo_coll.find(mongo_query).batch_size(mongo_batch_size)
 
# 定义更新函数
def update_docs(docs):
    for doc in docs:
        # 更新文档数据
        mongo_coll.update_one(
            {'_id': doc['_id']},
            {'$set': {'status': 'processed'}}
        )
 
# 分批次处理结果
num_threads = 4  # 定义线程数
docs_per_thread = 250  # 定义每个线程处理的文档数
threads = []
for i in range(num_threads):
    start_idx = i * docs_per_thread
    end_idx = (i+1) * docs_per_thread
    thread_docs = [doc for doc in mongo_results[start_idx:end_idx]]
    t = threading.Thread(target=update_docs, args=(thread_docs,))
    threads.append(t)
    t.start()
 
# 等待所有线程完成
for t in threads:
    t.join()

from pymongo import MongoClient
from concurrent.futures import ThreadPoolExecutor
 
# MongoDB 配置
mongo_uri = 'mongodb://localhost:27017/'
mongo_db_name = 'my_db'
mongo_collection_name = 'my_coll'
 
# 连接 MongoDB
mongo_client = MongoClient(mongo_uri)
mongo_db = mongo_client[mongo_db_name]
mongo_coll = mongo_db[mongo_collection_name]
 
# 查询 MongoDB
mongo_query = {}
mongo_batch_size = 1000
mongo_results = mongo_coll.find(mongo_query).batch_size(mongo_batch_size)
 
# 定义更新函数
def update_doc(doc):
    # 更新文档数据
    mongo_coll.update_one(
        {'_id': doc['_id']},
        {'$set': {'status': 'processed'}}
    )
 
# 使用线程池处理更新操作
num_threads = 4  # 定义线程数
with ThreadPoolExecutor(max_workers=num_threads) as executor:
    for doc in mongo_results:
        executor.submit(update_doc, doc)

from pymongo import MongoClient
from concurrent.futures import ThreadPoolExecutor
 
# MongoDB 配置
mongo_uri = 'mongodb://localhost:27017/'
mongo_db_name = 'my_db'
mongo_collection_name = 'my_coll'
 
# 连接 MongoDB
mongo_client = MongoClient(mongo_uri)
mongo_db = mongo_client[mongo_db_name]
mongo_coll = mongo_db[mongo_collection_name]
 
# 查询 MongoDB
mongo_query = {}
mongo_batch_size = 1000
mongo_results = mongo_coll.find(mongo_query).batch_size(mongo_batch_size)
 
# 定义更新函数
def update_doc(doc):
    # 更新文档数据
    mongo_coll.update_one(
        {'_id': doc['_id']},
        {'$set': {'status': 'processed'}}
    )
 
# 使用线程池处理更新操作
batch_size = 1000  # 定义每个批次的文档数量
num_threads = 4  # 定义并发线程数
with ThreadPoolExecutor(max_workers=num_threads) as executor:
    while True:
        batch_docs = list(mongo_results.next_n(batch_size))
        if not batch_docs:
            break
        for doc in batch_docs:
            executor.submit(update_doc, doc)