如何确保只有 4 个函数实例正在运行（使用多处理）？

Question

我正在使用 python 3.6.6 和 requests 和 bs4 包来下载和解析一些内容，现在我正在下载一些更大的文件 >1gb 并且只使用一个连接它相当慢所以我想加快它同时进行多个下载。

重要的代码片段：

def download(dir, link, name):
    r = requests.get(url, stream=True)
    with open(f'{path}/{filename}', 'wb') as f:
        for chunk in r.iter_content(chunk_size=1024):
            if chunk:
                f.write(chunk)

files = [{'link':'http://...','filename':'somename.7z'}]
download_dir= '~/Downloads'

for file in files:
    #do some things to check if file['link'] is valid and that the file dosen't already exist
    download(download_dir, file['link'], file['filename'])

我想做的是并行运行循环中的内容，确切地说是让循环中的内容同时运行 4 次。

我第一次尝试这样做是像这样使用multiprocessing.Pool.map：

def download(dir, link, name):
    r = requests.get(url, stream=True)
    with open(f'{path}/{filename}', 'wb') as f:
        for chunk in r.iter_content(chunk_size=1024):
            if chunk:
                f.write(chunk)

files = [{'link':'http://...','filename':'somename.7z'}]
download_dir= '~/Downloads'

datas = [{'file':f, 'dir':download_dir} for f in files]

worker(data)
    file = data['file']
    download_dir = data['dir']
    #do some things to check if file['link'] is valid and that the file dosen't already exist
    download(download_dir, file['link'], file['filename'])

pool = multiprocessing.Pool(4)
pool.map(worker, datas)

不幸的是，这不起作用，同时开始了超过 4 个下载，我假设它使用了 4 个线程，但是每次一个线程达到网络限制并且没有一个旧的更进一步，它只是启动了另一个 worker 实例.

为了强迫我的程序做我想做的事，我尝试了这种 hacky 方式：

def download(dir, link, name):
    r = requests.get(url, stream=True)
    with open(f'{path}/{filename}', 'wb') as f:
        for chunk in r.iter_content(chunk_size=1024):
            if chunk:
                f.write(chunk)

files = [{'link':'http://...','filename':'somename.7z'}]
download_dir= '~/Downloads'

worker(file, download_dir)
    #do some things to check if file['link'] is valid and that the file dosen't already exist
    download(download_dir, file['link'], file['filename'])

index = 0
while index < len(files):
    pool = multiprocessing.Pool(4)

    for _ in range(4):
        if index < len(files): #check exists cause I'm incrementing index in the inner for loop
            pool.apply_async(worker, (files[index], download_dir,))
        index += 1

    pool.close()
    pool.join()

但是pool.close() 并没有等待所有提交的任务完成，而是中止了下载，并且显然也不允许提交到池的任务在搁置后恢复。

这样做的正确方法是什么？

Answer 1

您的代码存在许多逻辑和样式错误。最大的一个是您为输入列表的每个元素创建一个新池！应该只有一个池。

if (index < len(urllist)) 始终为真，因为您在 while index < len(urlist) 内。

pool = multiprocessing.Pool(processes=size) 风格很差；您应该尽可能使用with（又名“上下文管理器”）。这将消除对pool.close() 和pool.join() 调用的需要。像这样：

with multiprocessing.Pool(processes=size) as pool:
    pool.apply_async(...)

数字 for 循环在 Python 中被认为是糟糕的风格，因为它们很容易避免。如果您只使用 for url in urllist: 或类似的，则不需要像 index = index + 1 这样的样板。

最后，您根本不需要循环，因为存在pool.map()（和pool.map_async()）并且可以在一个函数调用中完成所有操作。