99%人都不知道！一行Python代码竟然能实现并行！转载

原创

小哥 3年前 (2022-10-16) 阅读数 52 #大杂烩

Python 或多或少因程序并行化而臭名昭著。撇开技术问题不谈，例如线程和 GIL，我认为错误的指示是主要问题。通俗经典 Python 多线程、多进程的教程似乎更有偏见"重"。而且，他们常常挠痒痒，不去钻研日常工作中最有用的部分。

传统范例传统范例传统范例

简单搜索简单搜索"Python 多线程教程多线程教程"很容易看出，几乎所有的教程都给出了涉及类和队列的示例。

import os 
import PIL 

from multiprocessing import Pool 
from PIL import Image

SIZE = (75,75)
SAVE_DIRECTORY = thumbs

def get_image_paths(folder):
    return (os.path.join(folder, f) 
            for f in os.listdir(folder) 
            if jpeg in f)

def create_thumbnail(filename): 
    im = Image.open(filename)
    im.thumbnail(SIZE, Image.ANTIALIAS)
    base, fname = os.path.split(filename) 
    save_path = os.path.join(base, SAVE_DIRECTORY, fname)
    im.save(save_path)

if __name__ == __main__:
    folder = os.path.abspath(
        11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840)
    os.mkdir(os.path.join(folder, SAVE_DIRECTORY))

    images = get_image_paths(folder)

    pool = Pool()
    pool.map(creat_thumbnail, images)
    pool.close()
    pool.join()

哈，有点像哈，有点像 Java 不是吗？

我不是说使用Producer我不是说使用Producer/使用者模型处理多线程使用者模型处理多线程消费者模型处理多线程/多进程任务是错误的(事实上，该模型有其自身的用途)。只是在处理日常脚本任务时，我们可以使用更高效的模型。

问题是..。问题是……问题是。问题是……

首先，您需要一个样例类。(A)首先，你需要一个样板课。首先，您需要一个样例类。
其次，您需要一个队列来向其传递对象。
此外，您需要在通道的两端构建方法以帮助其工作(如果您想要双向通信或保存结果，还需要引入队列)。

worker 问题越多，问题就越多

沿着这些线，你现在需要一个 worker 线程的线程池。这是一个 IBM 经典教程中的例子--执行网络搜索时的多线程加速。

#Example2.py

A more realistic thread pool example 

import time 
import threading 
import Queue 
import urllib2 

class Consumer(threading.Thread): 
    def __init__(self, queue): 
        threading.Thread.__init__(self)
        self._queue = queue 

    def run(self):
        while True: 
            content = self._queue.get() 
            if isinstance(content, str) and content == quit:
                break
            response = urllib2.urlopen(content)
        print Bye byes!

def Producer():
    urls = [
        http://www.python.org, http://www.yahoo.com
        http://www.scala.org, http://www.google.com
        # etc.. 
    ]
    queue = Queue.Queue()
    worker_threads = build_worker_pool(queue, 4)
    start_time = time.time()

    # Add the urls to process
    for url in urls: 
        queue.put(url)  
    # Add the poison pillv
    for worker in worker_threads:
        queue.put(quit)
    for worker in worker_threads:
        worker.join()

    print Done! Time taken: {}.format(time.time() - start_time)

def build_worker_pool(queue, size):
    workers = []
    for _ in range(size):
        worker = Consumer(queue)
        worker.start() 
        workers.append(worker)
    return workers

if __name__ == __main__:
    Producer()

这段代码工作正常，但仔细看看我们需要做什么：构造不同的方法，跟踪一系列线程，并解决令人讨厌的死锁问题，我们需要做一系列 join 行动。而这只是个开始......

至此我们回顾了经典的多线程教程多线程教程，多少有些空洞不是吗？样板化而且易出错，这样事倍功半的风格显然不那么适合日常使用，好在我们还有更好的方法。

何不试试 map

map 这个小而精细的函数是 Python 程序并行化的关键。程序并行性的关键。程序并行化的关键。程序并行化的关键。map 源于 Lisp 这种类型的函数式编程语言。它可以通过一个序列实现两个函数之间的映射。

urls = [http://www.yahoo.com, http://www.reddit.com]
    results = map(urllib2.urlopen, urls)

上面的这两行代码将上面的两行代码将上面的这两行代码 urls 此序列中的每个元素都作为参数传递给 urlopen 方法并保存该方法中的所有结果，并将所有结果保存到 results 在这张单子上。结果大致相当于。

results = []
for url in urls: 
    results.append(urllib2.urlopen(url))

map 该函数单枪匹马地完成序列操作、参数传递、结果保存等一系列操作。

为什么这很重要？这是因为在正确的库的帮助下，map 可以很容易地实现操作的并行化。

在 Python 中有两个库有两个库包含有两个库包含 map 函数：multiprocessing 以及其不太为人所知的子库和不太为人所知的子库以及不太为人所知的子库和不太为人所知的子库 multiprocessing.dummy.

这里还有两个词。这里还有几个词。这里还有几个词。这里还有几个词。multiprocessing.dummy？mltiprocessing 库克隆的线程版本？这是虾吗？即使是在 multiprocessing 图书馆的官方文件中只有一句话描述了这个子库。这一描述翻译成人类术语，基本上是说:"嗯，有这样一件事，你知道它变成了."相信我，这个图书馆被严重低估了！相信我，这个图书馆被严重低估了！

dummy 是 multiprocessing 模块的完全克隆，唯一的区别是 multiprocessing 作用于过程，同时作用于过程，以及 dummy 模块作用于线程(因此还包括 Python (所有常见的多线程限制)。(所有常见的多线程限制)。(所有常见的多线程限制)。(所有常见的多线程限制)。
因此，替换这两个库非常容易。您可以针对 IO 集约化任务与集约化作业 CPU 选择不同库的密集任务。

动手尝试

使用以下两行代码引用包含并行化 map 函数库。函数库。函数库。

from multiprocessing import Pool
from multiprocessing.dummy import Pool as ThreadPool

实例化 Pool 对象：

pool = ThreadPool()

这个简单的语句代替这个简单的语句代替这个简单的短语代替这个简单的短语 example2.py 中 build worker pool 函数 7 要使用的代码行。它会生成一系列 worker 线程并完成初始化工作，将它们存储在变量中以便于访问。

Pool 对象有许多参数，这里我只需要关注它的第一个参数。processes. 此参数用于设置线程池中的线程数。它的默认值为当前计算机 CPU 的核数。

一般说来，执行一般，一般执行 CPU 在处理高强度任务时，调用的核心越多，速度就越快。但在处理网络密集型任务时，情况可能有点不可预测，因此明智的做法是尝试确定线程池的大小。

pool = ThreadPool(4) # Sets the pool size to 4

当线程数量太多时，切换线程可能会消耗比实际工作更多的时间。对于不同的作业，通过实验找到线程池大小的最佳值是一个好主意。

创建好 Pool 并行化的程序在对象之后出现。让我们来看看重写的 example2.py

import urllib2 
from multiprocessing.dummy import Pool as ThreadPool 

urls = [
    http://www.python.org, 
    http://www.python.org/about/,
    http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html,
    http://www.python.org/doc/,
    http://www.python.org/download/,
    http://www.python.org/getit/,
    http://www.python.org/community/,
    https://wiki.python.org/moin/,
    http://planet.python.org/,
    https://wiki.python.org/moin/LocalUserGroups,
    http://www.python.org/psf/,
    http://docs.python.org/devguide/,
    http://www.python.org/community/awards/
    # etc.. 
    ]

# Make the Pool of workers
pool = ThreadPool(4) 
# Open the urls in their own threads
# and return the results
results = pool.map(urllib2.urlopen, urls)
#close the pool and wait for the work to finish 
pool.close() 
pool.join()

唯一实际有效的代码是唯一可以正常工作的代码 4 线路，其中只有一条是关键的。线路，其中只有一条是关键的。线路，其中只有一个是关键的。线路，其中只有一个是关键的。map 函数轻松替换前一个覆盖函数轻松替换前一个覆盖函数 40 线程池的示例。为了让它更有趣，我计算了不同方法和不同线程池大小的时间消耗。

# results = [] 
# for url in urls:
#   result = urllib2.urlopen(url)
#   results.append(result)

# # ------- VERSUS ------- # 

# # ------- 4 Pool ------- # 
# pool = ThreadPool(4) 
# results = pool.map(urllib2.urlopen, urls)

# # ------- 8 Pool ------- # 

# pool = ThreadPool(8) 
# results = pool.map(urllib2.urlopen, urls)

# # ------- 13 Pool ------- # 

# pool = ThreadPool(13) 
# results = pool.map(urllib2.urlopen, urls)

结果：

#        Single thread:  14.4 Seconds 
#               4 Pool:   3.1 Seconds
#               8 Pool:   1.4 Seconds
#              13 Pool:   1.3 Seconds

很棒的结果，不是吗？这一结果还说明了为什么进行实验以确定线程池的大小很重要。在我的机器上，当线程池大小大于 9 好处将是非常有限的。好处是非常有限的。好处是有限的。

另一个真实的例子另一个真实的例子另一个真实的例子

生成数千张图片的缩略图生成数千张图片的缩略图生成数千张图片的缩略图
这是一个 CPU 任务密集型，非常适合并行化。

基本单流程版本基本单流程版本基本单流程版本

import os 
import PIL 

from multiprocessing import Pool 
from PIL import Image

SIZE = (75,75)
SAVE_DIRECTORY = thumbs

def get_image_paths(folder):
    return (os.path.join(folder, f) 
            for f in os.listdir(folder) 
            if jpeg in f)

def create_thumbnail(filename): 
    im = Image.open(filename)
    im.thumbnail(SIZE, Image.ANTIALIAS)
    base, fname = os.path.split(filename) 
    save_path = os.path.join(base, SAVE_DIRECTORY, fname)
    im.save(save_path)

if __name__ == __main__:
    folder = os.path.abspath(
        11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840)
    os.mkdir(os.path.join(folder, SAVE_DIRECTORY))

    images = get_image_paths(folder)

    for image in images:
        create_thumbnail(Image)

上述代码的主要工作是遍历传入文件夹中的图像文件，逐个生成缩略图，并将这些缩略图保存到特定文件夹。

这是我的机器，有这个程序要处理 6000 一幅画的成本是一幅画的成本一幅画的成本 27.9 秒。

如果我们使用如果我们使用如果我们正在使用 map 函数来替换函数而不是函数来替换函数而不是 for 循环：

import os 
import PIL 

from multiprocessing import Pool 
from PIL import Image

SIZE = (75,75)
SAVE_DIRECTORY = thumbs

def get_image_paths(folder):
    return (os.path.join(folder, f) 
            for f in os.listdir(folder) 
            if jpeg in f)

def create_thumbnail(filename): 
    im = Image.open(filename)
    im.thumbnail(SIZE, Image.ANTIALIAS)
    base, fname = os.path.split(filename) 
    save_path = os.path.join(base, SAVE_DIRECTORY, fname)
    im.save(save_path)

if __name__ == __main__:
    folder = os.path.abspath(
        11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840)
    os.mkdir(os.path.join(folder, SAVE_DIRECTORY))

    images = get_image_paths(folder)

    pool = Pool()
    pool.map(creat_thumbnail, images)
    pool.close()
    pool.join()

5.6 秒！

虽然只更改了几行代码，但我们显著提高了程序的执行速度。在生产环境中，我们可以向 CPU 集约化任务与集约化作业 IO 密集任务分别选择多进程和多线程库来进一步提高执行速度--这是解决死锁问题的秘诀。此外，由于 map 函数不支持手动线程管理，而是使关联的 debug 工作也变得非常容易。这项工作也变得非常容易。这项工作也变得非常容易。这项工作也变得非常容易。

到目前为止，我们已经用一行实现了(基本上)相同的事情 Python 实现了并行化。实现并行化。实现并行化。实现并行性。

译者：caspar

译文：https://segmentfault.com/a/1190000000414339

原文：https://medium.com/building-things-on-the-internet/40e9b2b36148

推荐阅读:
入门: 最完整的零基学习最全面的零基学习最完整的零基学习Python的问题  | 从零开始学习从零基础学习从零基础学习8个月的Python  | 实战项目 |学Python这是捷径这是捷径就是这条捷径
干货:爬行豆瓣短评，电影《后来的我们》 | 38年NBA最佳球员分析最佳球员分析 |   从万众期待到口碑惨败！唐探3令人失望  | 笑新一天图龙记笑新一天图龙集 | 谜语之王回答灯谜之王灯谜之王谜语之王 ｜用Python人山人海素描图人山人海素描图人山人海 Dishonor太火了，我用机器学习做了一个迷你推荐系统电影
趣味:弹球游戏  | 九宫格  | 漂亮的花 | 两百行Python日常酷跑游戏日常酷跑游戏日常酷跑游戏!
AI: 会写诗的机器人会写诗的机器人会写诗的机器人 | 给图片上色给图片上色给图片上色 | 预测收入 | 《耻辱》太火了，我用机器学习做了一部迷你推荐系统电影
小工具: Pdf转Word易于修复表单和水印！易于处理的表单和水印！易于修复表单和水印！简单的表格和水印！ | 一键把html将页面另存为网页另存为网页另存为pdf！|  再见PDF提款费！提款费！提款费！提款费用！ | 用90构建最强大的代码行构建最强大的代码行构建最强大的代码行PDF转换器，word、PPT、excel、markdown、html一键转换 | 制作一个固定的低成本机票提醒！制作一张别针的低价机票提醒！ ｜60代码行做了一个语音墙纸切换，天天见女士！

年度弹出文案年度弹出文案年度爆炸性文案

1). 卧槽！Pdf转Word用Python轻松搞定！
2).学Python闻起来好香！我用100一行代码做了一个网站，帮助人们做了一行代码，做了一个网站，帮助了人们做了一行代码，帮助了人们PS旅行图片赚鸡腿吃旅行图片赚鸡腿
3).第一次播放量过亿，火爆全网，我分析了《波妹》，发现了这些秘密
4). 80一行行代码！使用Python让救济金做正确的事做做的人做好事的人A梦分身
5).你必须掌握的东西你必须掌握20个python代码，简短而紧凑，永无止境的有用代码，简短而甜蜜，永无止境的有用的代码，简短而紧凑，永无止境的使用代码，简短而甜蜜，永无止境的用途
6). 30个Python古怪技能集古怪小贴士收藏古怪技能集
7). 我总结的80《菜鸟学习专页》《菜鸟学习专页》《菜鸟学习》Python精选干货.pdf》,都是干货
8). 再见Python！我要学Go了！2500词深度分析词深度分析词深度分析！
9).发现一只舔狗的福利！这Python爬虫神器太酷了，不能自动下载女孩的照片