建站
咨詢
售后
建站咨詢
新聞中心
眾所周知，Python的并行處理能力很不理想。我認(rèn)為如果不考慮線程和GIL的標(biāo)準(zhǔn)參數(shù)(它們大多是合法的)，其原因不是因?yàn)榧夹g(shù)不到位，而是我們的使用方法不恰當(dāng)。大多數(shù)關(guān)于Python線程和多進(jìn)程的教材雖然都很出色，但是內(nèi)容繁瑣冗長(zhǎng)。它們的確在開(kāi)篇鋪陳了許多有用信息，但往往都不會(huì)涉及真正能提高日常工作的部分。
成都創(chuàng)新互聯(lián)專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務(wù)，包含不限于網(wǎng)站設(shè)計(jì)制作、網(wǎng)站設(shè)計(jì)、鐵山港網(wǎng)絡(luò)推廣、微信小程序定制開(kāi)發(fā)、鐵山港網(wǎng)絡(luò)營(yíng)銷、鐵山港企業(yè)策劃、鐵山港品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運(yùn)營(yíng)等，從售前售中售后，我們都將竭誠(chéng)為您服務(wù)，您的肯定，是我們最大的嘉獎(jiǎng)；成都創(chuàng)新互聯(lián)為所有大學(xué)生創(chuàng)業(yè)者提供鐵山港建站搭建服務(wù)，24小時(shí)服務(wù)熱線：18980820575，官方網(wǎng)址：www.cdcxhl.com
經(jīng)典例子
DDG上以“Python threading tutorial (Python線程教程)”為關(guān)鍵字的熱門搜索結(jié)果表明：幾乎每篇文章中給出的例子都是相同的類+隊(duì)列。
事實(shí)上，它們就是以下這段使用producer/Consumer來(lái)處理線程/多進(jìn)程的代碼示例：
- #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()
唔…….感覺(jué)有點(diǎn)像Java。
我現(xiàn)在并不想說(shuō)明使用Producer / Consume來(lái)解決線程/多進(jìn)程的方法是錯(cuò)誤的——因?yàn)樗隙ㄕ_，而且在很多情況下它是最佳方法。但我不認(rèn)為這是平時(shí)寫(xiě)代碼的最佳選擇。
它的問(wèn)題所在(個(gè)人觀點(diǎn))
首先，你需要?jiǎng)?chuàng)建一個(gè)樣板式的鋪墊類。然后，你再創(chuàng)建一個(gè)隊(duì)列，通過(guò)其傳遞對(duì)象和監(jiān)管隊(duì)列的兩端來(lái)完成任務(wù)。(如果你想實(shí)現(xiàn)數(shù)據(jù)的交換或存儲(chǔ)，通常還涉及另一個(gè)隊(duì)列的參與)。
Worker越多，問(wèn)題越多。
接下來(lái)，你應(yīng)該會(huì)創(chuàng)建一個(gè)worker類的pool來(lái)提高Python的速度。下面是IBM tutorial給出的較好的方法。這也是程序員們?cè)诶枚嗑€程檢索web頁(yè)面時(shí)的常用方法。
- #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()
它的確能運(yùn)行，但是這些代碼多么復(fù)雜阿!它包括了初始化方法、線程跟蹤列表以及和我一樣容易在死鎖問(wèn)題上出錯(cuò)的人的噩夢(mèng)——大量的join語(yǔ)句。而這些還僅僅只是繁瑣的開(kāi)始!
我們目前為止都完成了什么?基本上什么都沒(méi)有。上面的代碼幾乎一直都只是在進(jìn)行傳遞。這是很基礎(chǔ)的方法，很容易出錯(cuò)(該死，我剛才忘了在隊(duì)列對(duì)象上還需要調(diào)用task_done()方法(但是我懶得修改了))，性價(jià)比很低。還好，我們還有更好的方法。
介紹：Map
Map是一個(gè)很棒的小功能，同時(shí)它也是Python并行代碼快速運(yùn)行的關(guān)鍵。給不熟悉的人講解一下吧，map是從函數(shù)語(yǔ)言Lisp來(lái)的。map函數(shù)能夠按序映射出另一個(gè)函數(shù)。例如
- urls = ['http://www.yahoo.com', 'http://www.reddit.com']
- results = map(urllib2.urlopen, urls)
這里調(diào)用urlopen方法來(lái)把調(diào)用結(jié)果全部按序返回并存儲(chǔ)到一個(gè)列表里。就像：
- results = []
- for url in urls:
- results.append(urllib2.urlopen(url))
Map按序處理這些迭代。調(diào)用這個(gè)函數(shù)，它就會(huì)返回給我們一個(gè)按序存儲(chǔ)著結(jié)果的簡(jiǎn)易列表。
為什么它這么厲害呢?因?yàn)橹灰辛撕线m的庫(kù)，map能使并行運(yùn)行得十分流暢!
有兩個(gè)能夠支持通過(guò)map函數(shù)來(lái)完成并行的庫(kù)：一個(gè)是multiprocessing，另一個(gè)是鮮為人知但功能強(qiáng)大的子文件：multiprocessing.dummy。
題外話：這個(gè)是什么?你從來(lái)沒(méi)聽(tīng)說(shuō)過(guò)dummy多進(jìn)程庫(kù)?我也是最近才知道的。它在多進(jìn)程的說(shuō)明文檔里面僅僅只被提到了一句。而且那一句就是大概讓你知道有這么個(gè)東西的存在。我敢說(shuō)，這樣幾近拋售的做法造成的后果是不堪設(shè)想的!
Dummy就是多進(jìn)程模塊的克隆文件。唯一不同的是，多進(jìn)程模塊使用的是進(jìn)程，而dummy則使用線程(當(dāng)然，它有所有Python常見(jiàn)的限制)。也就是說(shuō)，數(shù)據(jù)由一個(gè)傳遞給另一個(gè)。這能夠使得數(shù)據(jù)輕松的在這兩個(gè)之間進(jìn)行前進(jìn)和回躍，特別是對(duì)于探索性程序來(lái)說(shuō)十分有用，因?yàn)槟悴挥么_定框架調(diào)用到底是IO 還是CPU模式。
準(zhǔn)備開(kāi)始
要做到通過(guò)map函數(shù)來(lái)完成并行，你應(yīng)該先導(dǎo)入裝有它們的模塊：
- from multiprocessing import Pool
- from multiprocessing.dummy import Pool as ThreadPool
再初始化:
- pool = ThreadPool()
這簡(jiǎn)單的一句就能代替我們的build_worker_pool 函數(shù)在example2.py中的所有工作。換句話說(shuō)，它創(chuàng)建了許多有效的worker，啟動(dòng)它們來(lái)為接下來(lái)的工作做準(zhǔn)備，以及把它們存儲(chǔ)在不同的位置，方便使用。
Pool對(duì)象需要一些參數(shù)，但最重要的是：進(jìn)程。它決定pool中的worker數(shù)量。如果你不填的話，它就會(huì)默認(rèn)為你電腦的內(nèi)核數(shù)值。
如果你在CPU模式下使用多進(jìn)程pool，通常內(nèi)核數(shù)越大速度就越快(還有很多其它因素)。但是，當(dāng)進(jìn)行線程或者處理網(wǎng)絡(luò)綁定之類的工作時(shí)，情況會(huì)比較復(fù)雜所以應(yīng)該使用pool的準(zhǔn)確大小。
- pool = ThreadPool(4) # Sets the pool size to 4
如果你運(yùn)行過(guò)多線程，多線程間的切換將會(huì)浪費(fèi)許多時(shí)間，所以你最好耐心調(diào)試出最適合的任務(wù)數(shù)。
我們現(xiàn)在已經(jīng)創(chuàng)建了pool對(duì)象，馬上就能有簡(jiǎn)單的并行程序了，所以讓我們重新寫(xiě)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句還是簡(jiǎn)單的固定寫(xiě)法。調(diào)用map就能完成我們前面例子中40行的內(nèi)容!為了更形象地表明兩種方法的差異，我還分別給它們運(yùn)行的時(shí)間計(jì)時(shí)。
結(jié)果：
相當(dāng)出色!并且也表明了為什么要細(xì)心調(diào)試pool的大小。在這里，只要大于9，就能使其運(yùn)行速度加快。
實(shí)例2：
生成成千上萬(wàn)的縮略圖
我們?cè)贑PU模式下來(lái)完成吧!我工作中就經(jīng)常需要處理大量的圖像文件夾。其任務(wù)之一就是創(chuàng)建縮略圖。這在并行任務(wù)中已經(jīng)有很成熟的方法了。
基礎(chǔ)的單線程創(chuàng)建
- 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)
對(duì)于一個(gè)例子來(lái)說(shuō)，這是有點(diǎn)難，但本質(zhì)上，這就是向程序傳遞一個(gè)文件夾，然后將其中的所有圖片抓取出來(lái)，并最終在它們各自的目錄下創(chuàng)建和儲(chǔ)存縮略圖。
我的電腦處理大約6000張圖片用了27.9秒。
如果我們用并行調(diào)用map來(lái)代替for循環(huán)的話：
- 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秒!
對(duì)于只改變了幾行代碼而言，這是大大地提升了運(yùn)行速度。這個(gè)方法還能更快，只要你將cpu 和 io的任務(wù)分別用它們的進(jìn)程和線程來(lái)運(yùn)行——但也常造成死鎖。總之，綜合考慮到 map這個(gè)實(shí)用的功能，以及人為線程管理的缺失，我覺(jué)得這是一個(gè)美觀，可靠還容易debug的方法。
好了，文章結(jié)束了。一行完成并行任務(wù)。
當(dāng)前名稱：Python一行代碼完成并行任務(wù)
文章鏈接：http://www.fisionsoft.com.cn/article/coegccd.html
