服务粉丝

我们一直在努力
当前位置：首页 > 财经 >

一行 Python 代码竟然能实现并行！

日期： 来源：大侠学Python收集编辑：
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 不是吗？

我并不是说使用生产者/消费者模型处理多线程/多进程任务是错误的（事实上，这一模型自有其用武之地）。只是，处理日常脚本任务时我们可以使用更有效率的模型。

问题在于…

首先，你需要一个样板类； 
其次，你需要一个队列来传递对象； 
而且，你还需要在通道两端都构建相应的方法来协助其工作（如果需想要进行双向通信或是保存结果还需要再引入一个队列）。

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 中 buildworkerpool 函数 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

文章推荐

相关阅读

  • ChatGPT月活用户突破1亿

  • Datawhale干货 技术：ChatGPT，来源：机器之心继去年 12 月以最快速度（5天）突破百万用户之后，ChatGPT又创造了一个新的历史记录。在昨日宣布推出月费 20 美元的「ChatGPT Plus」会员

  • Kotlin中Lambda表达式及其简化

  • 在Kotlin编程中，lambda表达式的应用非常常见，这篇文章用来介绍一下怎样用lambda表达式定义函数，用lambda表达式进行函数传参，以及简化lambda表达式参考文档: https://developer.

  • 浅谈JS函数式编程

  • 高效搬砖——拆分、组合、复用我们不妨以一个栗子开始：这是Scott Sauyet所著《Favoring Curry》中一个在项目实战里遇到的问题。var data = { result: "SUCCESS", inte

  • 106. 2022的这一年

  • 说真的，2022马上就要过去了。。。又到了一年一度的回顾时间：乐观地来看，有了2022这一年垫底，以后还怕啥。。。这一年下来，怕是很多人已经忘了这个世界有多美了这次握手发生在整整

  • Java 判空新写法！干掉 if else 啦

  • 大家中午好，我是小咖，不管工作多忙，中午饭一定要吃哦。在文章的开头，先说下NPE问题，NPE问题就是，我们在开发中经常碰到的NullPointerException.假设我们有两个类，他们的UML类图如下

  • 日赚超亿却鲜有人知，全世界最高频的“印钞机”

  • 垄断性的科技，应用于金融市场，直接成为最高频的印钞机。这是一个鲜人知、但极度赚钱的商业帝国。在这个极其隐秘，但又具有高度竞争性的行当里，做的最好的这一家，每天都稳定地赚取

  • 9个技巧使你的Python代码更Pythonic！

  • "Beautiful is better than ugly." 这不仅是 "The Zen of Python" 的第一句话，也是所有Python开发者的信条。但如何区分漂亮和丑陋的代码？更重要的是，如何写出漂亮的 Python 代

  • IMREAL函数-可以冷但不可以凉凉~

  • 从最新的统计来看，Excel中目前有504个函数，其中常用的函数，只有几十个，那么剩下的那些不常用的函数中，有没有什么特别好用但是非常冷门的函数呢？

    且看一个案例




  • O365是函数式编程吗?!

  • 365版本及最近一批好用的新函数如下：关于学习，推荐加入学习圈，全部免费为什么今天发文这么晚？因为今天早上我还在更新系列教程！ 01:1834个函数，还有一个全部更新完毕！一起学起来吧！

  • Vlookup新用法，你的同事估计都不会

  • 关注我们并标星，带你成为Excel高手！今天跟大家分享下如何在带有合并单元格的表格中，使用Vlookup函数进行数据查询，这样的问题相信很多人都遇到过，在带有合并单元格的表格中使用函

聚合标签

中国 自己的 美国 都是 疫情 的人 的是 亿元 三星 手机 华为 这款 车型 万元 企业 小米 日本 俄罗斯 民警 项目

热门文章

  • “复活”半年后　京东拍拍二手杀入公益事业

  • 京东拍拍二手“复活”半年后，杀入公益事业，试图让企业捐的赠品、家庭闲置品变成实实在在的“爱心”。 把“闲置品”变爱心 6月12日，“益心一益·守护梦想每一步”2018年四

最新文章

  • 一行 Python 代码竟然能实现并行！

  • Python 在程序并行化方面多少有些声名狼藉。撇开技术上的问题，例如线程的实现和 GIL，我觉得错误的教学指导才是主要问题。常见的经典 Python 多线程、多进程教程多显得偏"重"

  • 最新1-5线城市完整名单

  • 第一财经·新一线城市研究所第一次构建《城市商业魅力排行榜》时，我们就特别强调了以现代城市商业文明的进程来指示中国城市实际繁荣程度的必要性。采用170个主流消费品牌的

  • 华平投资完成投资史上规模最大的基金募资

  • 近日，据地产资管界获悉，华平投资最新一期全球旗舰基金已经募资达154亿美元，接近其160亿美元的目标。该新基金成为华平投资历史上规模最大的基金。根据美国纽约州教师退休系统披