这是我想要实现的目标:
- 我有大约一百万个文件,需要解析并将已解析的内容附加到一个单独的文件中。
- 由于单个进程需要很长时间,因此此选项不可行。
- 不使用Python中的线程,因为它实际上只运行单个进程(由于GIL)。
- 因此使用multiprocessing模块。即生成4个子进程以利用所有原始核心能力:)
到目前为止还不错,现在我需要一个所有子进程都可以访问的共享对象。我正在使用multiprocessing模块中的Queues。此外,所有子进程都需要将其输出写入单个文件。可能需要使用锁定的地方。使用此设置运行时,我没有收到任何错误(因此父进程似乎正常),它只是停滞不前。当我按下ctrl-C时,我会看到一个回溯(每个子进程一个)。输出文件中也没有输出。以下是代码(请注意,如果没有多进程,一切都正常运行)-
import os
import glob
from multiprocessing import Process, Queue, Pool
data_file = open('out.txt', 'w+')
def worker(task_queue):
for file in iter(task_queue.get, 'STOP'):
data = mine_imdb_page(os.path.join(DATA_DIR, file))
if data:
data_file.write(repr(data)+'\n')
return
def main():
task_queue = Queue()
for file in glob.glob('*.csv'):
task_queue.put(file)
task_queue.put('STOP') # so that worker processes know when to stop
# this is the block of code that needs correction.
if multi_process:
# One way to spawn 4 processes
# pool = Pool(processes=4) #Start worker processes
# res = pool.apply_async(worker, [task_queue, data_file])
# But I chose to do it like this for now.
for i in range(4):
proc = Process(target=worker, args=[task_queue])
proc.start()
else: # single process mode is working fine!
worker(task_queue)
data_file.close()
return
我做错了什么?我也尝试在生成进程时将打开的文件对象传递给每个进程。但是没有效果。例如-
Process(target=worker, args=[task_queue, data_file])
。但这并没有改变任何事情。我觉得子进程由于某种原因无法写入文件。要么file_object
实例没有被复制(在生成时),要么其他问题......有人有想法吗?额外信息:还有没有办法保持一个持久的mysql_connection连接并将其传递给子进程?所以我在父进程中打开了一个mysql连接,所有子进程都可以访问打开的连接。基本上这相当于Python中的共享内存。这里有什么想法吗?