使用列表块时从同步到异步“处理”

Question

我通过Process 类使用多处理模块来执行一些不受 cpu 限制的任务，例如I/O 或 Web 请求。如果任务花费的时间太长，CPU 会达到 100% 的使用率（所有线程都在等待数据返回）。我怀疑异步执行解决方案，但我从未做过这样的事情。我正在使用的代码类似于以下内容，其中我有一个巨大的列表，每个进程都在一个块上工作。

您能在这个方向上提出建议吗？

提前致谢！！

import multiprocessing
def getData(urlsChunk, myQueue):

    for url in urlsChunk:
       fp = urllib.urlopen(url)
       try:
           data = fp.read()
           myQueue.put(data)
       finally:
           fp.close()

    return myQueue

manager = multiprocessing.Manager()
HUGEQ = manager.Queue()
urls = ['a huge list of url items']
chunksize = int(math.ceil(len(urls) / float(nprocs)))
for i in range(nprocs):
    p = Process(
                target = getData, # This is my worker
                args=(urls[chunksize * i:chunksize * (i + 1)],
                        MYQUEUE
                    )
                )
    processes.append(p)
    p.start()

for p in processes:
    p.join()

while True:
    try:
        MYQUEUEelem = MYQUEUE.get(block=False)
    except Empty:
        break
    else:
        'do something with the MYQUEUEelem'

Answer 1

使用multiprocessing.Pool，您的代码可以简化：

import multiprocessing

def getData(url):
    fp = urllib.urlopen(url)
    try:
        return fp.read()
    finally:
        fp.close()

if __name__ == '__main__':  # should protect the "entry point" of the program
    urls = ['a huge list of url items']
    pool = multiprocessing.Pool()
    for result in pool.imap(getData, urls, chunksize=10):
        # do something with the result