【发布时间】:2011-04-04 22:38:15
【问题描述】:
这就是我想要完成的 -
- 我有大约一百万个文件需要解析并将解析后的内容附加到单个文件中。
- 由于单个进程需要很长时间,因此此选项已失效。
- 不使用 Python 中的线程,因为它本质上是运行单个进程(由于 GIL)。
- 因此使用多处理模块。即产生 4 个子进程来利用所有原始核心功能:)
到目前为止一切顺利,现在我需要一个所有子进程都可以访问的共享对象。我正在使用多处理模块中的队列。此外,所有子流程都需要将其输出写入单个文件。我猜是使用锁的潜在场所。当我运行这个设置时,我没有收到任何错误(所以父进程看起来很好),它只是停止。当我按 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
我做错了什么?我还尝试在生成时将打开的 file_object 传递给每个进程。但是没有效果。例如-Process(target=worker, args=[task_queue, data_file])。但这并没有改变什么。我觉得子进程由于某种原因无法写入文件。 file_object 的实例没有被复制(在生成时)或其他一些怪癖......有人知道吗?
EXTRA: 还有有什么方法可以保持持久的 mysql_connection 打开并将其传递给 sub_processes?所以我在我的父进程中打开了一个 mysql 连接,并且我的所有子进程都应该可以访问打开的连接。基本上这相当于 python 中的 shared_memory 。这里有什么想法吗?
【问题讨论】:
-
如果您不写入文件而是进行打印,那么它可以工作吗? (在 Linux 上,我会使用 python script.py > out.dat 来防止屏幕泛滥)。
-
我认为 proc.start 是非阻塞的,所以你可能应该在某个地方等待,让进程有机会在你做 datafile.close() 之前做一些工作
-
data_file.close() 在最后完成。它应该在这里起作用吗?打印工作也很好。当我使用打印时,我在屏幕上看到输出......但我想使用文件。帮助!还有有什么方法可以保持持久的 mysql_connection 打开并将其传递给 sub_processes?
-
@extraneon: 很好,但是如果程序试图在一个关闭的文件上写入,应该引发一个异常。
-
你是在读mysql还是写mysql,还是两者兼而有之?
标签: python queue multiprocessing