用map函數(shù)來完成Python并行任務(wù)的簡單示例
眾所周知,Python的并行處理能力很不理想。我認(rèn)為如果不考慮線程和GIL的標(biāo)準(zhǔn)參數(shù)(它們大多是合法的),其原因不是因?yàn)榧夹g(shù)不到位,而是我們的使用方法不恰當(dāng)。大多數(shù)關(guān)于Python線程和多進(jìn)程的教材雖然都很出色,但是內(nèi)容繁瑣冗長。它們的確在開篇鋪陳了許多有用信息,但往往都不會涉及真正能提高日常工作的部分。
經(jīng)典例子
DDG上以“Python threading tutorial (Python線程教程)”為關(guān)鍵字的熱門搜索結(jié)果表明:幾乎每篇文章中給出的例子都是相同的類+隊列。
事實(shí)上,它們就是以下這段使用producer/Consumer來處理線程/多進(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()
唔…….感覺有點(diǎn)像Java。
我現(xiàn)在并不想說明使用Producer / Consume來解決線程/多進(jìn)程的方法是錯誤的——因?yàn)樗隙ㄕ_,而且在很多情況下它是最佳方法。但我不認(rèn)為這是平時寫代碼的最佳選擇。
它的問題所在(個人觀點(diǎn))
首先,你需要創(chuàng)建一個樣板式的鋪墊類。然后,你再創(chuàng)建一個隊列,通過其傳遞對象和監(jiān)管隊列的兩端來完成任務(wù)。(如果你想實(shí)現(xiàn)數(shù)據(jù)的交換或存儲,通常還涉及另一個隊列的參與)。
Worker越多,問題越多。
接下來,你應(yīng)該會創(chuàng)建一個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()
它的確能運(yùn)行,但是這些代碼多么復(fù)雜阿!它包括了初始化方法、線程跟蹤列表以及和我一樣容易在死鎖問題上出錯的人的噩夢——大量的join語句。而這些還僅僅只是繁瑣的開始!
我們目前為止都完成了什么?基本上什么都沒有。上面的代碼幾乎一直都只是在進(jìn)行傳遞。這是很基礎(chǔ)的方法,很容易出錯(該死,我剛才忘了在隊列對象上還需要調(diào)用task_done()方法(但是我懶得修改了)),性價比很低。還好,我們還有更好的方法。
介紹:Map
Map是一個很棒的小功能,同時它也是Python并行代碼快速運(yùn)行的關(guān)鍵。給不熟悉的人講解一下吧,map是從函數(shù)語言Lisp來的。map函數(shù)能夠按序映射出另一個函數(shù)。例如
urls = ['http://www.yahoo.com', 'http://www.reddit.com'] results = map(urllib2.urlopen, urls)
這里調(diào)用urlopen方法來把調(diào)用結(jié)果全部按序返回并存儲到一個列表里。就像:
results = [] for url in urls: results.append(urllib2.urlopen(url))
Map按序處理這些迭代。調(diào)用這個函數(shù),它就會返回給我們一個按序存儲著結(jié)果的簡易列表。
為什么它這么厲害呢?因?yàn)橹灰辛撕线m的庫,map能使并行運(yùn)行得十分流暢!
有兩個能夠支持通過map函數(shù)來完成并行的庫:一個是multiprocessing,另一個是鮮為人知但功能強(qiáng)大的子文件:multiprocessing.dummy。
題外話:這個是什么?你從來沒聽說過dummy多進(jìn)程庫?我也是最近才知道的。它在多進(jìn)程的說明文檔里面僅僅只被提到了一句。而且那一句就是大概讓你知道有這么個東西的存在。我敢說,這樣幾近拋售的做法造成的后果是不堪設(shè)想的!
Dummy就是多進(jìn)程模塊的克隆文件。唯一不同的是,多進(jìn)程模塊使用的是進(jìn)程,而dummy則使用線程(當(dāng)然,它有所有Python常見的限制)。也就是說,數(shù)據(jù)由一個傳遞給另一個。這能夠使得數(shù)據(jù)輕松的在這兩個之間進(jìn)行前進(jìn)和回躍,特別是對于探索性程序來說十分有用,因?yàn)槟悴挥么_定框架調(diào)用到底是IO 還是CPU模式。
準(zhǔn)備開始
要做到通過map函數(shù)來完成并行,你應(yīng)該先導(dǎo)入裝有它們的模塊:
from multiprocessing import Pool from multiprocessing.dummy import Pool as ThreadPool
再初始化:
pool = ThreadPool()
這簡單的一句就能代替我們的build_worker_pool 函數(shù)在example2.py中的所有工作。換句話說,它創(chuàng)建了許多有效的worker,啟動它們來為接下來的工作做準(zhǔn)備,以及把它們存儲在不同的位置,方便使用。
Pool對象需要一些參數(shù),但最重要的是:進(jìn)程。它決定pool中的worker數(shù)量。如果你不填的話,它就會默認(rèn)為你電腦的內(nèi)核數(shù)值。
如果你在CPU模式下使用多進(jìn)程pool,通常內(nèi)核數(shù)越大速度就越快(還有很多其它因素)。但是,當(dāng)進(jìn)行線程或者處理網(wǎng)絡(luò)綁定之類的工作時,情況會比較復(fù)雜所以應(yīng)該使用pool的準(zhǔn)確大小。
pool = ThreadPool(4) # Sets the pool size to 4
如果你運(yùn)行過多線程,多線程間的切換將會浪費(fèi)許多時間,所以你最好耐心調(diào)試出最適合的任務(wù)數(shù)。
我們現(xiàn)在已經(jīng)創(chuàng)建了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句還是簡單的固定寫法。調(diào)用map就能完成我們前面例子中40行的內(nèi)容!為了更形象地表明兩種方法的差異,我還分別給它們運(yùn)行的時間計時。
# 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)
結(jié)果:
# Single thread: 14.4 Seconds # 4 Pool: 3.1 Seconds # 8 Pool: 1.4 Seconds # 13 Pool: 1.3 Seconds
相當(dāng)出色!并且也表明了為什么要細(xì)心調(diào)試pool的大小。在這里,只要大于9,就能使其運(yùn)行速度加快。
實(shí)例2:
生成成千上萬的縮略圖
我們在CPU模式下來完成吧!我工作中就經(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)
對于一個例子來說,這是有點(diǎn)難,但本質(zhì)上,這就是向程序傳遞一個文件夾,然后將其中的所有圖片抓取出來,并最終在它們各自的目錄下創(chuàng)建和儲存縮略圖。
我的電腦處理大約6000張圖片用了27.9秒。
如果我們用并行調(diào)用map來代替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秒!
對于只改變了幾行代碼而言,這是大大地提升了運(yùn)行速度。這個方法還能更快,只要你將cpu 和 io的任務(wù)分別用它們的進(jìn)程和線程來運(yùn)行——但也常造成死鎖??傊C合考慮到 map這個實(shí)用的功能,以及人為線程管理的缺失,我覺得這是一個美觀,可靠還容易debug的方法。
好了,文章結(jié)束了。一行完成并行任務(wù)。
- Python中函數(shù)的返回值示例淺析
- Python 詳解基本語法_函數(shù)_返回值
- Python中用函數(shù)作為返回值和實(shí)現(xiàn)閉包的教程
- Python函數(shù)返回值實(shí)例分析
- python在回調(diào)函數(shù)中獲取返回值的方法
- Python的lambda匿名函數(shù)的簡單介紹
- 舉例講解Python的lambda語句聲明匿名函數(shù)的用法
- Python3匿名函數(shù)lambda介紹與使用示例
- Python中的filter()函數(shù)的用法
- 詳細(xì)分析python3的reduce函數(shù)
- Python map和reduce函數(shù)用法示例
- Python函數(shù)的返回值、匿名函數(shù)lambda、filter函數(shù)、map函數(shù)、reduce函數(shù)用法實(shí)例分析
相關(guān)文章
python讀取.mat文件的數(shù)據(jù)及實(shí)例代碼
這篇文章主要介紹了python讀取.mat文件的數(shù)據(jù)的方法,本文給大家介紹的非常詳細(xì),具有一定的參考借鑒價值 ,需要的朋友可以參考下2019-07-07基于Python+tkinter實(shí)現(xiàn)簡易計算器桌面軟件
tkinter是Python的標(biāo)準(zhǔn)GUI庫,對于初學(xué)者來說,它非常友好,因?yàn)樗峁┝舜罅康念A(yù)制部件,本文小編就來帶大家詳細(xì)一下如何利用tkinter制作一個簡易計算器吧2023-09-09python?配置uwsgi?啟動Django框架的詳細(xì)教程
這篇文章主要介紹了python?配置uwsgi?啟動Django框架,本文給大家講解的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2022-12-12ChatGLM-6B+LangChain環(huán)境部署與使用實(shí)戰(zhàn)
這篇文章主要介紹了ChatGLM-6B+LangChain環(huán)境部署與使用方法,結(jié)合實(shí)例形式詳細(xì)分析了ChatGLM-6B+LangChain環(huán)境部署相關(guān)步驟、實(shí)現(xiàn)方法與相關(guān)注意事項(xiàng),需要的朋友可以參考下2023-07-07