python多线程教程:并发与并行

2018-04-11

各种激烈的讨论都会经常的提到Python的多线程工作是多么的困难，指向被称为全局解释器锁阻碍了Python代码不能多线程同时执行。由于这个，如果你不是一个Python开发者以及你来自其他语言例如C++活着Java,线程模块表现出来的行为可能不是你所期望的那样。但是必须指出的是你仍可以写出并发或者并行的代码并使其表现出色，只要考虑全面。如果你还没有读到，我建议你看一下 Eqbal Quran’s 在Toptal博客上的article on concurrency and parallelism in Ruby。

在Python并发教程中，我们将会写出一个简单的Python脚本去下载来自Imgur排行靠前的图片，我们将从一个顺序下载的版本开始，或一次一个，作为先决条件，你将不得不注册一个应用在Imgur,如果你还没有创建一个Imgur账户，请先创建一个。

这个教程的脚本已经在Python3.4.2上测试通过，稍微作修改，就可以运行在Python2，urllib改动最大在这两个Python版本上。

开始在Python上使用多线程

让我们开始创建一个Python模块，并将其命名成”download.py”, 这个文件将会包含所有需要的函数去获取图片列表以及下载它们。我们将会分开这些函数为三个分离的函数。

get_links
download_link
setup_download_dir

第三个函数, “setup_download_dir”, 将会被用于创建下载保存目录如果不存在。

Imgur’s ApI要求HTTP请求头上携带客户端ID的”Authorization”标头，你可以从已经注册的Imgur的应用面板上查看客户端ID，请求返回结果会是JSON格式。我们可以使用Python的标准JSON库去解码。下载这些图片是一个非常简单的任务，只要我们获取到这些图片的URL以及将它们写入文件。

这个脚本看起来是这样的:

import json
import logging
import os
from pathlib import Path
from urllib.request import urlopen, Request

logger = logging.getLogger(__name__)

def get_links(client_id):
   headers = {'Authorization': 'Client-ID {}'.format(client_id)}
   req = Request('https://api.imgur.com/3/gallery/', headers=headers, method='GET')
   with urlopen(req) as resp:
       data = json.loads(resp.readall().decode('utf-8'))
   return map(lambda item: item['link'], data['data'])

def download_link(directory, link):
   logger.info('Downloading %s', link)
   download_path = directory / os.path.basename(link)
   with urlopen(link) as image, download_path.open('wb') as f:
       f.write(image.readall())

def setup_download_dir():
   download_dir = Path('images')
   if not download_dir.exists():
       download_dir.mkdir()
   return download_dir

接下来，我们将会写一个模块并使用这些函数去下载这些图片，一个接着一个，我们将会将其命名成”simple.py”,这个首先将会包含主函数，最初的Imgur图片下载器。这个模块将会从环境变量”IMGUR_CLIENT_ID”中接收Imgur的客户端ID，它将会在”setup_download_dir”中被调用去创建下载目标目录，最终，它会获取一个图片列表使用”get_links”函数，过滤所有的GIF和唱片URLs，然后使用”download_link”区下载以及保存每一张图片到硬盘中，这个”single.py”看起来如下:

import logging
import os
from time import time

from download import setup_download_dir, get_links, download_link

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logging.getLogger('requests').setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

def main():
   ts = time()
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   for link in links:
       download_link(download_dir, link)
   print('Took {}s'.format(time() - ts))

if __name__ == '__main__':
   main()

在我的笔记本上，这个脚本下载91张图片使用了19.4秒，请注意这些数字将因你的网络环境而异。但是我们我们想要下载更多的图片呢？或者900张，而不是90张，每一张图片平均使用0.2秒，900张图片将会使用大约3分钟，那么9000张图片将会使用掉30分钟。好消息是通过并发或并行介绍，我们可以显著的加速。

所有后续的代码示例将只会导入新的和特定的导入语句，为了方便，这些Python脚本可以在这个GitHub仓库找到。

使用多线程进行并发和并行

线程时最为熟知的方法实现Python并发与并行。线程通常由操作系统提供的功能。线程比进程更为轻量级，并且共用同一内存空间。

在我们的Python线程教程中，我们将会写一个新的模块代替”single.py”,这个模块将会创建一个8个线程的池，总共包含主线程一共9个线程。我选择8个线程，因为我的电脑有8个CPU以及每个工作线程运行在每个CPU核心上同时运行看起来会是一个好的选择，在练习上，这个数字是基于其他因素选择得应更为仔细，例如其他应用以及服务运行在同一台机器上。

这个和之前的一个非常相似，除了我们拥有了一个新类，DownloadWorker, 是Thread类下的一个子类，运行方法被重写了。它运行一个无限循环，在每一次的迭代上，它会调用”self.queue.get”去试着从线程安全的队列中获取一个URL，它会阻塞直到队列中有其他worker处理。一旦worker从队列中接收到一条，它接下来将会调用和之前脚本相同的”download_link”方法去下载图片以及保存到图片目录。在下载完成后，worker通知队列这个任务已经完成，这是非常重要的，因为队列跟踪有多少任务入队。然后调用”queue.join()”将会阻塞主线程如果workers没有通知它们已经完成了任务。

from queue import Queue
from threading import Thread

class DownloadWorker(Thread):
   def __init__(self, queue):
       Thread.__init__(self)
       self.queue = queue

   def run(self):
       while True:
           # Get the work from the queue and expand the tuple
           directory, link = self.queue.get()
           download_link(directory, link)
           self.queue.task_done()

def main():
   ts = time()
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   # Create a queue to communicate with the worker threads
   queue = Queue()
   # Create 8 worker threads
   for x in range(8):
       worker = DownloadWorker(queue)
       # Setting daemon to True will let the main thread exit even though the workers are blocking
       worker.daemon = True
       worker.start()
   # Put the tasks into the queue as a tuple
   for link in links:
       logger.info('Queueing {}'.format(link))
       queue.put((download_dir, link))
   # Causes the main thread to wait for the queue to finish processing all the tasks
   queue.join()
   print('Took {}'.format(time() - ts))

运行这个脚本在同一台机器上会更快的获取结果，只用了4.1秒!和之前的相比快了4.7倍。虽然这要快很多，但是值得提及的是由于GIL的原因同一时刻只有一个线程被执行，因此，这个代码是并发的但是不是并行的，相比之前快的原因在于这是一个IO绑定任务。处理器在下载这些图片时并不需要太耗费力气。大部分时间使用在等待网络上。这也是为什么线程可以提供大幅度的速度提升。处理器可以切换上下文在这些线程无论其中的哪一个准备去做一些任务。在Python或者任务其他带有GIL解释性语言使用线程模块实际上可能导致性能下降，如果你的代码是CPU绑定任务，例如解压文件，使用多线程模块将会使结果更慢，对于CPU绑定的任务来讲和真正的并行执行，我们将会使用multiprocessing模块。

事实上参考Python的实现，CPython，拥有GIL，不是所有的Python解释器都是这样。例如，IronPython,一个是使用.NET 框架实现的Python解释器，并不会有GIL，诸如Jython,基于Java实现的，你可以在这找到一系列Python解释器。

使用多进程

多进程模块比多线程模块更容易使用，因为我们不需要添加类似于线程示例的类。唯一改变的是主函数。

使用多进程我们可以创建一个进程池，使用提供的map方法，我们可以将列表中的URLs放入池中。将会产生8个新进程并使用每个进程并行下载这些图片，这是真正的并行。但它也带来的成本。脚本的整个内存被复制到各个子进程中。在这个简单的例子中，这不是什么大不了的事情，但是它可能很容易产生严重开销。

from functools import partial
from multiprocessing.pool import Pool

def main():
   ts = time()
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   download = partial(download_link, download_dir)
   with Pool(8) as p:
       p.map(download, links)
   print('Took {}s'.format(time() - ts))

分配给多个worker

虽然线程和进程脚本都非常适合运行在个人电脑上，如果你想运行在不同的机器上你应该怎么做呢？或者向上扩展增加更多的CPU。对于长期运行的web项目来讲会是一个很好的用例。如果你有长时间运行任务，你不想在同一台机器上启动一堆需要运行的应用程序在多进程或多线程上。这将会降低你的应用性能对所有的你的用户，最好能够在另外一台或多台上运行这些任务。

对于这种任务一个出色的Python库是RQ,一个非常简单但是又非常功能强大的库，你首先需要确定函数和它的参数,pickles将会被调用，并将其加入到Redis列表中，声明工作排队是第一步，但是他并不会做任何事情，我们至少需要一个worker去监听这个工作队列。

第一步我们需要安装和运行Redis服务在你的个人电脑上，或者拥有访问Redis服务的权限，然后，只会在原有的代码基础上稍作改动，我们首先创建一个RQ队列实例并且连接到Redis服务上通过redis-py library,然后，代替刚才调用的”download_link”方法，我们使用“q.enqueue(download_link, download_dir, link)”。这个队列将函数作为它的第一个参数，然后其他参数将会带入到这个函数直到这个任务被执行。

最后一步我们需要启动这些worker,RQ提供了一系列的脚本去运行这些workers在默认的队列上，只需要在终端是那个执行”rqworker”然后它会启动一个worker监听这个默认的队列，请确保你目前的工作目录和这些脚本是在同一目录中，如果你想监听不同的队列，你可以运行 “rqworker queue_name”然后它就会监听倍命名的队列。关于RQ最出色的地方在于只要你连接到Redis,你可以运行任意多个worker在不同的机器上。这是非常容易扩展的。这个是RQ的版本:

from redis import Redis
from rq import Queue

def main():
   client_id = os.getenv('IMGUR_CLIENT_ID')
   if not client_id:
       raise Exception("Couldn't find IMGUR_CLIENT_ID environment variable!")
   download_dir = setup_download_dir()
   links = [l for l in get_links(client_id) if l.endswith('.jpg')]
   q = Queue(connection=Redis(host='localhost', port=6379))
   for link in links:
       q.enqueue(download_link, download_dir, link)

然而，RQ不是唯一的Python工作队列解决方式，RQ是简单使用并且非常好的覆盖简单用例，如果需要其他高级的特性，其他工作队列，如Celery可以被使用。

结论

如果你的代码是IO绑定的，multiprocessing和multithreading都适合你，多进程比多线程更容易使用，但是会占用更多的内存，如果你的代码是CPU绑定的，那么multiprocessing可能是更优的选择。尤其目标机器拥有多个CPU，对于web应用，如果想要扩展worker,RQ会是更好的选择。

此片文章翻译自:beginners-guide-to-concurrency-and-parallelism-in-python.