Python中大文件的并发下载和处理

Question

我有一组URL指向大文件（例如压缩档案）需要下载，然后再进行处理（例如解压缩）。由于下载和处理都需要较长时间，而且处理过程对磁盘IO要求很高，因此我希望每次只运行一个任务。由于两个任务需要的时间差不多，并且不会互相竞争资源，因此我想在上一个任务被处理时下载下一个文件。

这是生产者-消费者问题的变种。

情况类似于读取和处理图像或下载大量文件，但我的下载调用目前还不能使用多进程，而且两个任务所需时间相同。

以下是一个虚拟示例，其中下载和处理都是阻塞的：

import time
import posixpath

def download(urls):
    for url in urls:
        time.sleep(3)  # this is the download (more like 1000s) 
        yield posixpath.basename(url)

def process(fname):
    time.sleep(2)  # this is the processing part (more like 600s)

urls = ['a', 'b', 'c']
for fname in download(urls):
    process(fname)
    print(fname)

我该如何使这两个任务并发执行？我可以使用 yield 或者 yield from 以更加巧妙的方式, 或者与 deque 结合使用吗？或者必须使用 asyncio 和 Future？

- j08lue

---

可能是复制自：https://dev59.com/Y2Qo5IYBdhLWcg3wR9rD - Fallenreaper

不完全正确。那个是关于多个并发下载的。我只需要一次下载一个文件，消费者应该知道每个新文件。 - j08lue

Answer 1

一年后，我们实际上正在使用Python 3的asyncio和aiohttp。

- j08lue

Answer 2

我会简单地使用 threading.Thread(target=process, args=(fname,)) 并启动一个新线程进行处理。

但在此之前，需要结束上一个处理线程：

t = None
for fname in download(urls):
    if t is not None: # wait for last processing thread to end
        t.join()
    t = threading.Thread(target=process, args=(fname,))
    t.start()
    print('[i] thread started for %s' % fname)

请参阅https://docs.python.org/3/library/threading.html

- Loïc

---

很酷，没错，那应该可行而且相当简单。 - j08lue

请仅返回翻译后的文本：即使我盲目编码并未测试，它也应该可以正常工作。如果有任何问题，请告知我以便我修复我的答案。 - Loïc

我相当确定这是一个可行的答案，但问题是我的代码运行在一些应用程序（QGIS）内部，似乎不喜欢我使用Python的threading（它会崩溃）。我必须调查细节，但基于协程的解决方案可能更安全。 - j08lue

我认为在for循环结束时缺少t.join()。在else子句中添加它。 - Laurent LAPORTE

我已经导入了线程。我将尝试在应用程序之外使用这种解决方案，并让您知道它是否有效。@LaurentLAPORTE 似乎是对的。你怎么看，@Loïc？ - j08lue

我不认为它丢失了，也许在最后一次循环中没有join()，尽管程序不需要特别完成线程。 - Loïc