python - 多个进程从 Python 中的中央源读取的快速方法？

Question

我正在寻找一种让多个进程(在 multiprocessing.Pool() 中)从中央数据源读取的快速方法。目前我有一个文件被读入队列(使用 multiprocessing.Manager().Queue())，然后启动工作池并从该队列读取其进程。它工作正常，但是当我处理多个 GB 大小的文件时，这就成了一个问题，因为托管队列比常规 python 队列慢 ~7 倍。

我认为这是因为管理器处于一个单独的进程中，它必须通过套接字而不是直接与内存进行通信。

这是我正在使用的代码(get_records 函数只是从文件中读取每条记录的字节流并返回它)

from multiprocessing import Manager
manager = Manager()
mgr_q = manager.Queue()
map(mgr_q.put, get_records(f))

那么也许有更好的方法来处理这个问题？

---

以下是关于将我的一个数据文件 (~3GB) 读取为各种数据类型的速度的一些统计数据:

读入常规 python 列表。速率为 229.377 MB/秒

l = []
map(l.append, get_records(f))

读入常规队列。速率为 74.035 MB/秒

import Queue
q = Queue.Queue()
map(q.put, get_records(f))

读入 multiprocessing.queues 队列。速率为 67.718 MB/秒

from multiprocessing import Queue
mq = Queue()
map(mq.put, get_records(f))

最后读入托管队列。速率为 9.568 MB/秒

from multiprocessing import Manager
manager = Manager()
mgr_q = manager.Queue()
map(mgr_q.put, get_records(f))

速率计算方式为rate = duration/filesize/1024/1024

Answer 1

如果您只是对文件进行读取，那么同时读取多个进程是安全的。不传递队列中的数据，只传递偏移量和计数。比在 worker 身上做:

f.seek(offset)
get_records(f, count)