如何同时下载多个文件并在python中加入它们？

Question

我在远程服务器上有一些拆分文件.

我已经尝试逐个下载它们并加入它们.但这需要很多时间.我用谷歌搜索,发现同时下载可能会加快速度.该脚本在Python上.

我的伪是这样的:

url1 = something
url2 = something
url3 = something

data1 = download(url1)
data2 = download(url2)
data3 = download(url3)

wait for all download to complete
join all data and save

任何人都可以指出我可以同时加载文件的方向,并等到它们完成.

我试过创建一个类.但我再也无法弄清楚如何等到一切完成.

我对线程和队列功能更感兴趣,我可以在我的平台中导入它们.

我已尝试使用Thread和Queue,并在此站点上找到示例.这是代码pastebin.com/KkiMLTqR.但它不会等待或永远等待......不确定

Answer 1

有两种方法可以同时做事.或者,真的,2-3/4左右:

• 多线程
  • 或者多个进程,特别是如果"事物"需要大量的CPU能力
  • 或coroutines或greenlets,特别是如果有成千上万的"东西"
  • 或上述之一的池
• 事件循环(手动编码)
  • 或者混合greenlet /事件循环系统gevent.

如果您有1000个URL,则可能不希望同时执行1000个请求.例如,Web浏览器通常一次只执行8个请求.游泳池是一次只做8件事的好方法,所以让我们这样做.

而且,既然你一次只做8件事,那些事情主要受I/O限制,线程就是完美的.

我将实现它futures.(如果您使用的是Python 2.x或3.0-3.1,则需要安装backport,.futures)

import concurrent.futures

urls = ['http://example.com/foo', 
        'http://example.com/bar']

with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
    result = b''.join(executor.map(download, urls))

with open('output_file', 'wb') as f:
    f.write(result)

当然你需要编写这个download函数,但如果你一次只做这些函数,那就完全一样了.

例如,使用urlopen(如果您使用的是Python 2.x,请使用urllib2而不是urllib.request):

def download(url):
    with urllib.request.urlopen(url) as f:
        return f.read()

如果你想学习如何自己构建一个线程池执行器,那么源实际上非常简单,并且multiprocessing.pool是stdlib中另一个很好的例子.

但是,这两者都有很多过多的代码(处理弱引用以提高内存使用率,干净地关闭,提供不同的等待结果的方式,正确地传播异常等等)可能会妨碍你.

如果你环顾PyPI和ActiveState,你会发现更简单的设计threadpool,你可能会发现更容易理解.

但这是最简单的可连接线程池:

class ThreadPool(object):
    def __init__(self, max_workers):
        self.queue = queue.Queue()
        self.workers = [threading.Thread(target=self._worker) for _ in range(max_workers)]
    def start(self):
        for worker in self.workers:
            worker.start()
    def stop(self):
        for _ in range(self.workers):
            self.queue.put(None)
        for worker in self.workers:
            worker.join()
    def submit(self, job):
        self.queue.put(job)
    def _worker(self):
        while True:
            job = self.queue.get()
            if job is None:
                break
            job()

当然,简单实现的缺点是使用起来并不友好concurrent.futures.ThreadPoolExecutor:

urls = ['http://example.com/foo', 
        'http://example.com/bar']
results = [list() for _ in urls]
results_lock = threading.Lock()

def download(url, i):
    with urllib.request.urlopen(url) as f:
        result = f.read()
    with results_lock:
        results[i] = url

pool = ThreadPool(max_workers=8)
pool.start()
for i, url in enumerate(urls):
    pool.submit(functools.partial(download, url, i))
pool.stop()

result = b''.join(results)

with open('output_file', 'wb') as f:
    f.write(result)