一行代码完成并行任务

时间 2019-11-09

标签一行代码完成并行任务繁體版

原文原文链接

众所周知，Python的并行处理能力很不理想。我认为若是不考虑线程和GIL的标准参数（它们大可能是合法的），其缘由不是由于技术不到位，而是咱们的使用方法不恰当。大多数关于Python线程和多进程的教材虽然都很出色，可是内容繁琐冗长。它们的确在开篇铺陈了许多有用信息，但每每都不会涉及真正能提升平常工做的部分。html

经典例子

DDG上以“Python threading tutorial （Python线程教程）”为关键字的热门搜索结果代表：几乎每篇文章中给出的例子都是相同的类+队列。python

事实上，它们就是如下这段使用producer/Consumer来处理线程/多进程的代码示例：程序员

#Example.py
'''
Standard Producer/Consumer Threading Pattern
'''

import time 
import threading 
import Queue 

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

    def run(self):
        while True: 
            # queue.get() blocks the current thread until 
            # an item is retrieved. 
            msg = self._queue.get() 
            # Checks if the current message is 
            # the "Poison Pill"
            if isinstance(msg, str) and msg == 'quit':
                # if so, exists the loop
                break
            # "Processes" (or in our case, prints) the queue item   
            print "I'm a thread, and I received %s!!" % msg
        # Always be friendly! 
        print 'Bye byes!'

def Producer():
    # Queue is used to share items between
    # the threads.
    queue = Queue.Queue()

    # Create an instance of the worker
    worker = Consumer(queue)
    # start calls the internal run() method to 
    # kick off the thread
    worker.start() 

    # variable to keep track of when we started
    start_time = time.time() 
    # While under 5 seconds.. 
    while time.time() - start_time < 5: 
        # "Produce" a piece of work and stick it in 
        # the queue for the Consumer to process
        queue.put('something at %s' % time.time())
        # Sleep a bit just to avoid an absurd number of messages
        time.sleep(1)

    # This the "poison pill" method of killing a thread. 
    queue.put('quit')
    # wait for the thread to close down
    worker.join()

if __name__ == '__main__':
    Producer()

唔…….感受有点像Java。web

我如今并不想说明使用Producer / Consume来解决线程/多进程的方法是错误的——由于它确定正确，并且在不少状况下它是最佳方法。但我不认为这是平时写代码的最佳选择。segmentfault

它的问题所在（我的观点）

首先，你须要建立一个样板式的铺垫类。而后，你再建立一个队列，经过其传递对象和监管队列的两端来完成任务。（若是你想实现数据的交换或存储，一般还涉及另外一个队列的参与）。网络

Worker越多，问题越多。

接下来，你应该会建立一个worker类的pool来提升Python的速度。下面是IBM tutorial给出的较好的方法。这也是程序员们在利用多线程检索web页面时的经常使用方法。多线程

#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语句。而这些还仅仅只是繁琐的开始！app

咱们目前为止都完成了什么？基本上什么都没有。上面的代码几乎一直都只是在进行传递。这是很基础的方法，很容易出错（该死，我刚才忘了在队列对象上还须要调用task_done()方法（可是我懒得修改了）），性价比很低。还好，咱们还有更好的方法。框架

介绍：Map

Map是一个很棒的小功能，同时它也是Python并行代码快速运行的关键。给不熟悉的人讲解一下吧，map是从函数语言Lisp来的。map函数可以按序映射出另外一个函数。例如ide

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

这里调用urlopen方法来把调用结果所有按序返回并存储到一个列表里。就像：

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

Map按序处理这些迭代。调用这个函数，它就会返回给咱们一个按序存储着结果的简易列表。

为何它这么厉害呢？由于只要有了合适的库，map能使并行运行得十分流畅！

有两个可以支持经过map函数来完成并行的库：一个是multiprocessing，另外一个是不为人知但功能强大的子文件：multiprocessing.dummy。

题外话：这个是什么？你历来没据说过dummy多进程库？我也是最近才知道的。它在多进程的说明文档里面仅仅只被提到了一句。并且那一句就是大概让你知道有这么个东西的存在。我敢说，这样几近抛售的作法形成的后果是不堪设想的！

Dummy就是多进程模块的克隆文件。惟一不一样的是，多进程模块使用的是进程，而dummy则使用线程（固然，它有全部Python常见的限制）。也就是说，数据由一个传递给另外一个。这可以使得数据轻松的在这两个之间进行前进和回跃，特别是对于探索性程序来讲十分有用，由于你不用肯定框架调用究竟是IO 仍是CPU模式。

准备开始

要作到经过map函数来完成并行，你应该先导入装有它们的模块：

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

再初始化:

pool = ThreadPool()

这简单的一句就能代替咱们的build_worker_pool 函数在example2.py中的全部工做。换句话说，它建立了许多有效的worker，启动它们来为接下来的工做作准备，以及把它们存储在不一样的位置，方便使用。

Pool对象须要一些参数，但最重要的是：进程。它决定pool中的worker数量。若是你不填的话，它就会默认为你电脑的内核数值。

若是你在CPU模式下使用多进程pool，一般内核数越大速度就越快（还有不少其它因素）。可是，当进行线程或者处理网络绑定之类的工做时，状况会比较复杂因此应该使用pool的准确大小。

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

若是你运行过多线程，多线程间的切换将会浪费许多时间，因此你最好耐心调试出最适合的任务数。

咱们如今已经建立了pool对象，立刻就能有简单的并行程序了，因此让咱们从新写example2.py中的url opener吧！

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行就完成了全部的工做。其中3句仍是简单的固定写法。调用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

至关出色！而且也代表了为何要细心调试pool的大小。在这里，只要大于9，就能使其运行速度加快。

实例2：

生成成千上万的缩略图

咱们在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(create_thumbnail,images)
        pool.close()
        pool.join()

5.6秒！

对于只改变了几行代码而言，这是大大地提高了运行速度。这个方法还能更快，只要你将cpu 和 io的任务分别用它们的进程和线程来运行——但也常形成死锁。总之，综合考虑到 map这个实用的功能，以及人为线程管理的缺失，我以为这是一个美观，可靠还容易debug的方法。

好了，文章结束了。一行完成并行任务。

原文：Parallelism in one line
转载自：伯乐在线 - colleen__chen