Python:多处理代码非常慢

Question

我使用pymongo从mongodb一次性提取.8百万条记录(这是一次性过程)并执行一些操作超过它 。

我的代码如下所示。

    proc = []
    for rec in cursor: # cursor has .8 million rows 
            print cnt
            cnt = cnt + 1
            url =  rec['urlk']
            mkptid = rec['mkptid']
            cii = rec['cii']

            #self.process_single_layer(url, mkptid, cii)


            proc = Process(target=self.process_single_layer, args=(url, mkptid, cii))
            procs.append(proc)
            proc.start()

             # complete the processes
    for proc in procs:
        proc.join()

process_single_layer 是一个基本上从云端下载 url 并存储在本地的函数。

现在的问题是下载过程很慢，因为它必须点击一个网址。由于处理 1k 行的记录非常庞大，因此需要 6 分钟。

为了减少我想要实现多处理的时间。但很难看出与上面的代码有什么区别。

请建议我如何提高这种情况下的性能。

Answer 1

首先，您需要计算文件中的所有行，然后生成固定数量的进程(理想情况下与处理器核心的数量匹配)，您通过队列(每个进程一个)向其提供多个进程行数等于除法total_number_of_rows/number_of_cores。这种方法背后的想法是在多个进程之间分割这些行的处理，从而实现并行性。

动态找出核心数量的方法是:

import multiprocessing as mp
cores_count = mp.cpu_count()

通过避免初始行计数可以实现的一个轻微改进是通过创建队列列表来循环添加行，然后在其上应用循环迭代器。

完整示例:

import queue
import multiprocessing as mp
import itertools as itools

cores_count = mp.cpu_count()


def dosomething(q):

    while True:

        try:
            row = q.get(timeout=5)
        except queue.Empty:
            break

    # ..do some processing here with the row

    pass

if __name__ == '__main__':
    processes
    queues = []

    # spawn the processes
    for i in range(cores_count):
        q = mp.Queue()
        queues.append(q)
        proc = Process(target=dosomething, args=(q,))
        processes.append(proc)

    queues_cycle = itools.cycle(queues)
    for row in cursor:
        q = next(queues_cycle)
        q.put(row)

    # do the join after spawning all the processes
    for p in processes:
        p.join()