Python多处理，不能腌制thread.lock（pymongo）

Posted 2023-04-15

【中文标题】Python多处理，不能腌制thread.lock（pymongo）

【英文标题】：Python multiprocessing, can't pickle thread.lock (pymongo)

【发布时间】：2016-12-10 02:29:04

【问题描述】：

我有一个具有以下方法的类：

def get_add_new_links(self, max_num_links):
    self.get_links_m2(max_num_links)
    processes = mp.cpu_count()
    pool = mp.Pool(processes=processes)
    func = partial(worker, self)
    with open(os.path.join(self.report_path, "links.txt"), "r") as f:
        reports = pool.map(func, f.readlines())
    pool.close()
    pool.join()

其中get_links_m2 是另一种创建文件“links.txt”的方法。工人是：

def worker(obje, link):
    doc, rep = obje.get_info_m2(link)
    obje.add_new_active(doc, sure_not_exists=True)
    return rep

方法get_info_m2访问链接并提取一些信息。 add_new_active 方法将信息添加到 MongoDB。

我的代码可能有什么问题？当我运行它时，我得到了这个错误（和回溯）：

文件“controller.py”，第 234 行，在 get_add_new_links 中

reports = pool.map(func, f.readlines())   File "/home/vladimir/anaconda3/lib/python3.5/multiprocessing/pool.py", line

260，在地图中

return self._map_async(func, iterable, mapstar, chunksize).get()   File "/home/vladimir/anaconda3/lib/python3.5/multiprocessing/pool.py",

第 608 行，在获取中

raise self._value   File "/home/vladimir/anaconda3/lib/python3.5/multiprocessing/pool.py", line

385，在_handle_tasks中

put(task)   File "/home/vladimir/anaconda3/lib/python3.5/multiprocessing/connection.py",

第 206 行，发送中

self._send_bytes(ForkingPickler.dumps(obj))   File "/home/vladimir/anaconda3/lib/python3.5/multiprocessing/reduction.py",

第 50 行，转储中

cls(buf, protocol).dump(obj) TypeError: can't pickle _thread.lock objects

【参考方案1】：

如the docs中所述：

永远不要这样做：

client = pymongo.MongoClient()

# Each child process attempts to copy a global MongoClient
# created in the parent process. Never do this.
def func():
  db = client.mydb
  # Do something with db.

proc = multiprocessing.Process(target=func)
proc.start()

相反，必须在工作函数内部初始化客户端。

【讨论】：

多线程不是这样，它工作正常：with ThreadPool(64) as pool: collection = pymongo.MogoClient()["db"]["coll"] r = pool.map(do_something, collection) pool.join()