批评Python的讨论经常讨论如何难以使用Python进行多线程工作, 指向所谓的全局解释器锁(亲切地称为 吉尔)，它可以防止多个Python代码线程同时运行. 因此, Python多线程模块的行为并不完全像您所期望的那样，如果您不是 Python开发人员 而你来自其他语言，如c++或Java. 必须明确的是，仍然可以用Python编写并发或并行运行的代码，并在最终性能上产生明显的差异, 只要考虑到某些事情. 如果你还没有读过，我建议你看一下平等古兰经 关于Ruby中的并发和并行的文章 在Toptal工程博客上.

演示Python中的并发性, 我们将编写一个小脚本，从Imgur下载最受欢迎的图片. 我们将从一个按顺序下载图像或一次下载一个图像的版本开始. 作为先决条件，你必须注册 Imgur上的一个应用程序. 如果您还没有Imgur帐户，请先创建一个.

这些Python多线程示例中的脚本已在Python 3中进行了测试.6.4. 做了一些改变, 它们也应该与Python 2一起运行- urllib是这两个Python版本之间变化最大的地方.

Python多线程入门

让我们从创建Python模块开始，命名为 下载.py. 该文件将包含获取和下载图像列表所需的所有函数. 我们将这些功能分成三个独立的函数:

• get_links
• 下载_link
• setup_下载_dir

第三个函数， setup_下载_dir，如果下载目标目录不存在，则使用该参数创建下载目标目录.

Imgur的API需要HTTP请求来承受 授权 头与客户端ID. 您可以从您在Imgur上注册的应用程序的仪表板中找到这个客户端ID, 响应将是JSON编码. 我们可以使用Python的标准JSON库来解码它. 下载图像是一个更简单的任务, 因为您所要做的就是通过URL获取图像并将其写入文件.

脚本看起来是这样的:

进口json
导入日志
进口操作系统
从路径lib导入路径
从urllib.请求导入ur循环en

日志记录.getlog (__name__)

类型= {'图像/jpeg'， '图像/png'}


def get_links (client_id):
    headers = {'授权': 'Client-ID {}'.格式(client_id)}
    req =请求('http://api.imgur.'， headers=headers, method='GET')
    以urlopen(req)为例:
        Data = json.负载(分别地.read ().解码(utf - 8))
    如果item中有type，而type中有item['type']，则返回data['data']中的item['link']


Def 下载_link(目录，链接):
    Download_路径 = directory / OS.路径.basename(链接)
    使用urlopen(链接)作为图像，下载路径.打开('wb')作为f:
        f.写(图片.read ())
    日志记录器.info('已下载%s'，链接)


def setup_下载_dir ():
    下载_dir = Path('图片')
    如果不是，下载_dir.存在():
        下载_dir.mkdir ()
    返回下载_dir

下一个, 我们需要编写一个模块，使用这些函数来下载图像, 一个接一个. 我们给它命名 单.py. 这将包含我们第一个简单版本的Imgur图像下载器的主要功能. 该模块将在环境变量中检索Imgur客户端ID IMGUR_CLIENT_ID. 它将调用 setup_下载_dir 创建下载目标目录. 最后，它将使用 get_links 函数，过滤掉所有GIF和相册的url，然后使用 下载_link 下载并将每个映像保存到磁盘. 下面是 单.py 看起来像:

导入日志
进口操作系统
从时间导入时间

从下载导入setup_下载_dir, get_links, 下载_link

日志记录.basicConfig(=日志级别.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
日志记录.getlog (__name__)

def main ():
    Ts = time()
    Client_id = OS.采用“IMGUR_CLIENT_ID”)
    如果不是client_id:
        无法找到IMGUR_CLIENT_ID环境变量!")
    Download_dir = setup_下载_dir()
    链接= get_links(client_id)
    连结中的连结:
        下载_link (下载_dir链接)
    日志记录.info('用时%s秒'，time() - ts)

如果__name__ == '__main__':
    main ()

在我的笔记本电脑上，这个脚本花了19分钟.4秒下载91张图片. 请注意，这些数字可能会根据您所使用的网络而有所不同. 19.4秒不是很长，但是如果我们想下载更多的图片呢? 也许是900张图片，而不是90张. 平均值为0.每张图片2秒，900张图片大约需要3分钟. 拍摄9000张照片需要30分钟. 好消息是，通过引入并发性或并行性，我们可以显著加快这一速度.

所有后续的代码示例将只显示新的和特定于这些示例的import语句. 为方便起见，所有这些Python脚本都可以在 这个GitHub存储库.

Python中的并行性和并发性:多线程示例

线程是在Python中实现并行性和并发性的最著名的方法之一. 线程化通常是操作系统提供的一个特性. 线程比进程轻，并且共享相同的内存空间.

在这个Python多线程示例中，我们将编写一个新模块来替换 单.py. 该模块将创建一个包含八个线程的池, 包括主线程在内总共有9个线程. 我选择了8个工作线程，因为我的计算机有8个CPU核心，每个核心一个工作线程似乎是一次运行多少个线程的好数字. 在实践中, 这个数字的选择要根据其他因素仔细得多, 例如在同一台机器上运行的其他应用程序和服务.

这和前一个几乎是一样的，除了我们现在有一个新的类， DownloadWorker它是蟒蛇的后代 线程 class. run方法已被重写，它将运行一个无限循环. 在每次迭代中，它调用 自我.队列.get () 尝试从线程安全队列中获取URL. 它会阻塞，直到队列中有一个项目供工人处理. 一旦工作线程从队列中接收到一个项目，它就会调用该项目 下载_link 方法，该方法在前面的脚本中用于将图像下载到图片目录. 下载完成后，工人向队列发出任务完成的信号. 这一点非常重要，因为队列会跟踪进入队列的任务数量. 呼唤 队列.加入() 如果工作线程没有发出完成任务的信号，会永远阻塞主线程吗.

导入日志
进口操作系统
从队列导入队列
从线程导入线程
从时间导入时间

从下载导入setup_下载_dir, get_links, 下载_link


日志记录.basicConfig(=日志级别.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

日志记录.getlog (__name__)


类DownloadWorker(线程):

    Def __init__(自我, 队列):
        线程.__init__(自我)
        自我.队列=排队

    def运行(自我):
        而真正的:
            #从队列中获取工作并扩展元组
            目录，link = 自我.队列.get ()
            试一试:
                下载_link(目录,链接)
            最后:
                自我.队列.task_done ()


def main ():
    Ts = time()
    Client_id = OS.采用“IMGUR_CLIENT_ID”)
    如果不是client_id:
        无法找到IMGUR_CLIENT_ID环境变量!")
    Download_dir = setup_下载_dir()
    链接= get_links(client_id)
    #创建一个队列与工作线程通信
    队列=排队 ()
    创建8个工作线程
    对于范围(8)内的x:
        工人 = DownloadWorker(队列)
        #设置daemon为True将使主线程退出，即使工作线程阻塞
        工人.daemon = True
        工人.start ()
    #将任务以元组的形式放入队列
    连结中的连结:
        日志记录器.信息(“排队{}”.格式(链接))
        队列.((下载_dir链接))
    导致主线程等待队列处理完所有任务
    队列.加入()
    日志记录.info('占用%s'， time() - s)

如果__name__ == '__main__':
    main ()

在之前使用的同一台机器上运行这个Python多线程示例脚本的下载时间为4.1秒! 这是4.比之前的例子快7倍. 这要快得多, 值得一提的是，由于吉尔，在整个过程中一次只执行一个线程. 因此，这段代码是并发的，但不是并行的. 它仍然更快的原因是因为这是一个IO绑定任务. 处理器在下载这些图像时几乎不会出汗, 大部分时间都花在等待网络上. 这就是为什么Python多线程可以提供很大的速度提升. 当其中一个线程准备好执行某些工作时，处理器可以在线程之间切换. 在Python或任何其他带有吉尔的解释性语言中使用线程模块实际上会导致性能降低. 如果您的代码正在执行CPU绑定任务，例如解压缩gzip文件，则使用 线程 模块将导致较慢的执行时间. 对于CPU受限的任务和真正的并行执行，我们可以使用多处理模块.

而 事实上的 参考Python实现- cpython -有一个吉尔，这并不是所有Python实现都是如此. 例如，IronPython，一个使用 .. NET框架没有吉尔，基于java的实现Jython也没有吉尔. 您可以找到工作Python实现的列表 在这里.

Python 多处理:生成多个进程

Python多处理模块比线程模块更容易使用, 因为我们不需要像Python多线程示例那样添加一个类. 我们需要做的唯一更改是在主函数中.

为了使用多个进程，我们创建了一个多进程 池. 使用它提供的map方法, 我们将把url列表传递给池, 这将依次产生8个新进程，并使用每个进程并行下载图像. 这是真正的并行，但它是有代价的. 脚本的整个内存被复制到生成的每个子进程中. 在这个简单的例子中, 这没什么大不了的, 但它很容易成为重要程序的严重开销.

导入日志
进口操作系统
从functools导入partial
从多处理.pool import pool
从时间导入时间

从下载导入setup_下载_dir, get_links, 下载_link


日志记录.basicConfig(=日志级别.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
日志记录.getlog(请求).setLevel(日志记录.关键)
日志记录.getlog (__name__)


def main ():
    Ts = time()
    Client_id = OS.采用“IMGUR_CLIENT_ID”)
    如果不是client_id:
        无法找到IMGUR_CLIENT_ID环境变量!")
    Download_dir = setup_下载_dir()
    链接= get_links(client_id)
    下载= partial(下载_link, 下载_dir)
    设池(4)为p:
        p.地图(下载链接)
    日志记录.info('用时%s秒'，time() - ts)


如果__name__ == '__main__':
    main ()

分配给多个工作人员

而Python多线程和多处理模块对于在个人计算机上运行的脚本来说是非常好的, 如果您希望在另一台机器上完成工作，应该怎么做, 或者您需要扩展到一台机器上的CPU无法处理的范围? 一个很好的用例是web应用程序的长时间后端任务. 如果您有一些长时间运行的任务, 您不希望在同一台机器上启动一堆需要运行其余应用程序代码的子进程或线程. 这将降低应用程序对所有用户的性能. 如果能够在另一台机器或许多其他机器上运行这些作业，那就太棒了.

用于此任务的一个很好的Python库是 RQ，一个非常简单但功能强大的库. 首先使用库将函数及其参数排队. 这 泡菜 函数调用表示，然后将其附加到 复述, 列表. 将作业加入队列是第一步，但还不会做任何事情. 我们还需要至少一个工人来监听该作业队列.

第一步是在计算机上安装并运行复述,服务器, 或者访问正在运行的复述,服务器. 在此之后，只需要对现有代码进行一些小的更改. 我们首先创建一个RQ 队列的实例，并将复述,服务器的实例传递给它 redis-py图书馆. 然后，不要只是调用我们的 下载_link 方法，我们调用 q.En队列 (下载_link, 下载_dir, link). en队列方法接受一个函数作为其第一个参数, 然后，在实际执行作业时，将任何其他参数或关键字参数传递给该函数.

我们需要做的最后一步是启动一些工人. RQ提供了一个方便的脚本来在默认队列上运行工人. 你就跑 rq工人 在终端窗口中，它将启动一个监听默认队列的工作线程. 请确保您当前的工作目录与脚本所在的目录相同. 如果您想要收听不同的队列，您可以运行 rq工人 队列_name 它会监听那个命名队列. RQ的伟大之处在于，只要你能连接到复述,, you can run as many 工人s as you like on as many different machines as you like; t在这里fore, 随着应用程序的增长，扩展是非常容易的. 以下是RQ版本的源代码:

导入日志
进口操作系统

导入redis

从rq导入队列

从下载导入setup_下载_dir, get_links, 下载_link


日志记录.basicConfig(=日志级别.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
日志记录.getlog(请求).setLevel(日志记录.关键)
日志记录.getlog (__name__)


def main ():
    Client_id = OS.采用“IMGUR_CLIENT_ID”)
    如果不是client_id:
        无法找到IMGUR_CLIENT_ID环境变量!")
    Download_dir = setup_下载_dir()
    链接= get_links(client_id)
    q =队列(连接=复述,(主机='localhost'，端口=6379))
    连结中的连结:
        q.En队列 (下载_link, 下载_dir, link)

如果__name__ == '__main__':
    main ()

然而，RQ并不是唯一的Python作业队列解决方案. RQ易于使用，并且非常好地覆盖了简单的用例, 但如果需要更高级的选择, 其他Python 3队列解决方案(例如 芹菜)可以使用.

多处理和. Python中的多线程

如果您的代码是IO绑定的，那么Python中的多处理和多线程都可以为您工作. Python多处理比线程化更容易实现，但它的内存开销更高. 如果您的代码是CPU限制的, 多处理很可能是更好的选择—特别是如果目标机器有多个内核或cpu. web应用程序, 当您需要跨多台机器扩展工作时, RQ会更适合你.

---

 

更新

改进Python 3中的并发性.2 + 并发.期货

Python 3以来的新功能.原文中没有提到的第二个问题是 并发.期货 包. 这个包提供了另一种在Python中使用并行性和并发性的方法.

在原文中, 我提到过，Python多处理模块比线程模块更容易放入现有代码中. 这是因为Python 3的线程模块需要子类化 线程 类，还创建了 队列 用于监视工作的线程.

使用一个 并发.期货.线程池Executor 使得Python线程示例代码几乎与多处理模块相同.

导入日志
进口操作系统
从并发.导入线程池Executor
从functools导入partial
从时间导入时间

从下载导入setup_下载_dir, get_links, 下载_link

日志记录.basicConfig(=日志级别.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

日志记录.getlog (__name__)


def main ():
    Client_id = OS.采用“IMGUR_CLIENT_ID”)
    如果不是client_id:
        无法找到IMGUR_CLIENT_ID环境变量!")
    Download_dir = setup_下载_dir()
    链接= get_links(client_id)

    #通过将遗嘱执行人放置在with块中，遗嘱执行人关闭方法
    #将被称为清理线程.
    # 
    #默认情况下，执行器将工人的数量设置为5倍
    # cpu.
    使用线程池Executor()作为执行器:

        创建一个新的部分应用函数来存储目录
        #参数.
        # 
        #这允许下载_link函数，通常需要两个参数
        #参数与map函数一起工作，该函数需要a的函数
        #单参数.
        Fn = partial(下载_link, 下载_dir)

        #使用可迭代链接上的线程并发执行fn. 的
        # timeout适用于整个进程，而不是单个调用，所以请下载
        所有图像必须在30秒内完成.
        遗嘱执行人.Map (fn, links, timeout=30)


如果__name__ == '__main__':
    main ()

现在我们已经用Python下载了所有这些图像 线程池Executor，我们可以使用它们来测试cpu绑定的任务. 我们可以在单线程中创建所有图像的缩略图版本, 单进程脚本，然后测试基于多进程的解决方案.

我们要用 枕头 库来处理图像的大小调整.

这是我们的初始脚本.

导入日志
从路径lib导入路径
从时间导入时间

从PIL导入图像

日志记录.basicConfig(=日志级别.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

日志记录.getlog (__name__)


Def create_thumbnail(大小，路径):
    """
    创建与图像名称相同的图像缩略图
    _thumbnail附加在扩展名之前.  E.g.:

    >>> create_thumbnail((128, 128), '图像.jpg”)

    以图像_thumbnail的名称创建一个新的缩略图.jpg

    :param size:图像的宽度和高度的元组
    :param 路径:镜像文件的路径
    返回:没有
    """
    图像 =图像.打开(路径)
    图像.缩略图(尺寸)
    路径 =路径(路径)
    Name = 路径.茎+ '_thumbnail' +路径.后缀
    Thumbnail_路径 = 路径.with_name(名字)
    图像.保存(thumbnail_路径)


def main ():
    Ts = time()
    for 图像_路径 in Path('图片').iterdir ():
        Create_thumbnail ((128, 128)， 图像_路径)
    日志记录.info('占用%s'， time() - s)


如果__name__ == '__main__':
    main ()

中的路径进行迭代 图片 文件夹，并为每个路径运行create_thumbnail函数. 这个函数使用枕头打开图像, 创建缩略图, 保存新的, 较小的图像，与原始图像名称相同，但带有 _thumbnail 附加到名称后.

在160张总共3600万张的图片上运行这个脚本需要2个小时.32秒. 看看能不能用a加快速度 Process池Executor.

导入日志
从路径lib导入路径
从时间导入时间
从functools导入partial

从并发.导入Process池Executor

从PIL导入图像

日志记录.basicConfig(=日志级别.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

日志记录.getlog (__name__)


Def create_thumbnail(大小，路径):
    """
    创建与图像名称相同的图像缩略图
    _thumbnail附加在扩展名之前. E.g.:

    >>> create_thumbnail((128, 128), '图像.jpg”)

    以图像_thumbnail的名称创建一个新的缩略图.jpg

    :param size:图像的宽度和高度的元组
    :param 路径:镜像文件的路径
    返回:没有
    """
    路径 =路径(路径)
    Name = 路径.茎+ '_thumbnail' +路径.后缀
    Thumbnail_路径 = 路径.with_name(名字)
    图像 =图像.打开(路径)
    图像.缩略图(尺寸)
    图像.保存(thumbnail_路径)


def main ():
    Ts = time()
    #部分应用create_thumbnail方法，设置大小为128x128
    #并返回单个参数的函数.
    Thumbnail_128 = partial(create_thumbnail， (128, 128))

    #在with块中创建执行器，这样当block被关闭时就会被调用
    #退出.
    使用Process池Executor()作为执行器:
        遗嘱执行人.地图(thumbnail_128路径('图像').iterdir ())
    日志记录.info('占用%s'， time() - s)


如果__name__ == '__main__':
    main ()

的 create_thumbnail 方法与上一个脚本相同. 主要区别在于创建 Process池Executor. 执行程序的 map 方法用于并行创建缩略图. 默认情况下， Process池Executor 为每个CPU创建一个子进程. 在同样的160张图片上运行这个脚本花费了1.05年seconds-2.快2倍!

Async/Await (Python 3.5 +)

在原文章的评论中，请求最多的项目之一是使用Python 3的示例 asyncio 模块. 和其他例子相比, 有一些新的Python语法对大多数人来说可能是新的，也有一些新的概念. 一个不幸的额外复杂性层是由Python的内置 urllib 模块不是异步的. 我们需要使用异步HTTP库来获得asyncio的全部好处. 对此，我们将使用 aiohttp.

让我们直接进入代码，后面会有更详细的解释.

进口asyncio
导入日志
进口操作系统
从时间导入时间

进口aiohttp

从下载导入setup_下载_dir, get_links

日志记录.basicConfig(=日志级别.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
日志记录.getlog (__name__)


Async def async_下载_link(会话，目录，链接):
    """
    我们在其他示例中使用的下载_link方法的Async版本.
    :参数session: aiohttp ClientSession
    :param directory:保存下载文件的目录
    :param link:要下载的链接的url
    返回:
    """
    Download_路径 = directory / OS.路径.basename(链接)
    带会话的异步.获取(链接)作为响应:
        与下载_路径.打开('wb')作为f:
            而真正的:
                # 等待暂停执行，直到从流中读取1024(或更少)字节
                Chunk =等待响应.内容.阅读(1024)
                如果不是chunk:
                    #我们完成了文件的读取，跳出而循环
                    打破
                f.写(块)
    日志记录器.info('已下载%s'，链接)


# Main现在是一个协程
Async def ():
    Client_id = OS.采用“IMGUR_CLIENT_ID”)
    如果不是client_id:
        无法找到IMGUR_CLIENT_ID环境变量!")
    Download_dir = setup_下载_dir()
    我们使用会话来利用tcp的keep-alive
    设置3秒的读取和连接超时. 默认为5分钟
    与aiohttp异步.ClientSession(conn_timeout=3, read_timeout=3)作为session:
        Tasks = [(async_下载_link(session, 下载_dir, l)) for lin get_links(client_id)]
        # gather聚合所有任务并在事件循环中调度它们
        等待asyncio.收集(*任务,return_exceptions = True)


如果__name__ == '__main__':
    Ts = time()
    创建asyncio事件循环
    循环= asyncio.get_event_循环 ()
    试一试:
        循环.run_until_complete (main ())
    最后:
        #即使有异常也关闭循环
        循环.close ()
    日志记录器.info('耗时%s秒完成'，time() - ts)

这里有很多东西需要解开. 让我们从程序的主要入口点开始. 我们使用asyncio模块做的第一件新事情是获取事件循环. 事件循环处理所有异步代码. 然后循环运行，直到完成并通过 main 函数. 在main的定义中有一个新的语法: 异步def. 你还会注意到 等待 和 与异步.

async/等待语法是在 PEP492. 的 异步def 语法将函数标记为 协同程序. 在内部，协程是基于Python生成器的，但并不是完全相同的东西. 协程返回一个协程对象，类似于生成器返回生成器对象的方式. 有了协程后，就可以使用 等待 表达式. 当协程调用时 等待，则暂停协程的执行，直到可等待对象完成. 此挂起允许在协程挂起“等待”某些结果时完成其他工作. 在一般情况下, 这个结果将是某种I/O，比如数据库请求，或者在我们的例子中是HTTP请求.

的 下载_link 功能必须有很大的改变. 以前，我们依靠 urllib 为我们做读图的首要工作. 现在，为了使我们的方法能够在异步编程范例中正常工作，我们引入了 而 循环，每次读取映像的块，并在等待I/O完成时暂停执行. 这允许事件循环遍历下载不同的图像，因为每个图像在下载期间都有新的可用数据.

应该有一种——最好只有一种——显而易见的方法

而 Python之禅 告诉我们应该有一个显而易见的方法去做某件事, 在Python中有很多方法可以将并发引入我们的程序中. 选择的最佳方法将取决于您的具体用例. 与线程或多处理相比，异步范式更适合于高并发工作负载(如web服务器), 但它要求您的代码(和依赖项)是异步的，以便充分受益.

希望本文中的Python多线程示例(以及更新)将为您指明正确的方向，以便您知道如果需要在程序中引入并发性，应该在Python标准库中查找什么地方.