自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

Python一行代碼完成并行任務(wù)

作者：佚名
開(kāi)發(fā) 后端
眾所周知，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ì)涉及真正能提高日常工作的部分。

眾所周知，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ì)涉及真正能提高日常工作的部分。

經(jīng)典例子

DDG上以“Python threading tutorial (Python線程教程)”為關(guān)鍵字的熱門(mén)搜索結(jié)果表明：幾乎每篇文章中給出的例子都是相同的類+隊(duì)列。

事實(shí)上，它們就是以下這段使用producer/Consumer來(lái)處理線程/多進(jìn)程的代碼示例：

  1. #Example.py 
  2.  
  3. ''' 
  4.  
  5.     Standard Producer/Consumer Threading Pattern 
  6.  
  7. ''' 
  8.  
  9.   
  10.  
  11. import time 
  12.  
  13. import threading 
  14.  
  15. import Queue 
  16.  
  17.   
  18.  
  19. class Consumer(threading.Thread): 
  20.  
  21. def __init__(self, queue): 
  22.  
  23.     threading.Thread.__init__(self) 
  24.  
  25.     self._queue = queue 
  26.  
  27.   
  28.  
  29. def run(self): 
  30.  
  31.     while True: 
  32.  
  33.         # queue.get() blocks the current thread until 
  34.  
  35.         # an item is retrieved. 
  36.  
  37.         msg = self._queue.get() 
  38.  
  39.         # Checks if the current message is 
  40.  
  41.         # the "Poison Pill" 
  42.  
  43.         if isinstance(msg, str) and msg == 'quit': 
  44.  
  45.             # if so, exists the loop 
  46.  
  47.             break 
  48.  
  49.         # "Processes" (or in our case, prints) the queue item 
  50.  
  51.         print "I'm a thread, and I received %s!!" % msg 
  52.  
  53.         # Always be friendly! 
  54.  
  55.     print 'Bye byes!' 
  56.  
  57.   
  58.  
  59. def Producer(): 
  60.  
  61.     # Queue is used to share items between 
  62.  
  63.     # the threads. 
  64.  
  65.     queue = Queue.Queue() 
  66.  
  67.   
  68.  
  69.     # Create an instance of the worker 
  70.  
  71.     worker = Consumer(queue) 
  72.  
  73.     # start calls the internal run() method to 
  74.  
  75.     # kick off the thread 
  76.  
  77.     worker.start() 
  78.  
  79.   
  80.  
  81.     # variable to keep track of when we started 
  82.  
  83.     start_time = time.time() 
  84.  
  85.     # While under 5 seconds.. 
  86.  
  87.     while time.time() - start_time < 5: 
  88.  
  89.         # "Produce" a piece of work and stick it in 
  90.  
  91.         # the queue for the Consumer to process 
  92.  
  93.         queue.put('something at %s' % time.time()) 
  94.  
  95.     # Sleep a bit just to avoid an absurd number of messages 
  96.  
  97.     time.sleep(1) 
  98.  
  99.   
  100.  
  101.     # This the "poison pill" method of killing a thread. 
  102.  
  103.     queue.put('quit') 
  104.  
  105.     # wait for the thread to close down 
  106.  
  107.     worker.join() 
  108.  
  109.   
  110.  
  111. if __name__ == '__main__': 
  112.  
  113. 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í)的常用方法。

  1. #Example2.py 
  2.  
  3. """ 
  4.  
  5. A more realistic thread pool example 
  6.  
  7. """ 
  8.  
  9.   
  10.  
  11. import time 
  12.  
  13. import threading 
  14.  
  15. import Queue 
  16.  
  17. import urllib2 
  18.  
  19.   
  20.  
  21. class Consumer(threading.Thread): 
  22.  
  23.     def __init__(self, queue): 
  24.  
  25.         threading.Thread.__init__(self) 
  26.  
  27.         self._queue = queue 
  28.  
  29.   
  30.  
  31.     def run(self): 
  32.  
  33.         while True: 
  34.  
  35.             content = self._queue.get() 
  36.  
  37.             if isinstance(content, str) and content == "quit": 
  38.  
  39.                 break 
  40.  
  41.             response = urllib2.urlopen(content) 
  42.  
  43.        print "Bye byes!" 
  44.  
  45.   
  46.  
  47. def Producer(): 
  48.  
  49.     urls = [ 
  50.  
  51.          "http://www.python.org&#039;, &#039;http://www.yahoo.com", 
  52.  
  53.         "http://www.scala.org&#039;, &#039;http://www.google.com", 
  54.  
  55.     # etc.. 
  56.  
  57.     ] 
  58.  
  59.     queue = Queue.Queue() 
  60.  
  61.     worker_threads = build_worker_pool(queue, 4) 
  62.  
  63.     start_time = time.time() 
  64.  
  65.   
  66.  
  67.     # Add the urls to process 
  68.  
  69.     for url in urls: 
  70.  
  71.         queue.put(url)   
  72.  
  73.     # Add the poison pillv 
  74.  
  75.     for worker in worker_threads: 
  76.  
  77.         queue.put("quit") 
  78.  
  79.     for worker in worker_threads: 
  80.  
  81.         worker.join() 
  82.  
  83.   
  84.  
  85.     print "Done! Time taken: {}".format(time.time() - start_time) 
  86.  
  87.   
  88.  
  89. def build_worker_pool(queue, size): 
  90.  
  91.     workers = [] 
  92.  
  93.     for _ in range(size): 
  94.  
  95.         worker = Consumer(queue) 
  96.  
  97.         worker.start() 
  98.  
  99.         workers.append(worker) 
  100.  
  101.     return workers 
  102.  
  103.   
  104.  
  105. if __name__ == &#039;__main__&#039;: 
  106.  
  107.     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ù)。例如

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

這里調(diào)用urlopen方法來(lái)把調(diào)用結(jié)果全部按序返回并存儲(chǔ)到一個(gè)列表里。就像：

  1. results = [] 
  2.  
  3. for url in urls: 
  4.  
  5. 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)入裝有它們的模塊：

  1. from multiprocessing import Pool 
  2.  
  3. from multiprocessing.dummy import Pool as ThreadPool  

再初始化:

  1. 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)確大小。

  1. 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吧!

  1. import urllib2 
  2.  
  3. from multiprocessing.dummy import Pool as ThreadPool 
  4.  
  5.   
  6.  
  7. urls = [ 
  8.  
  9. 'http://www.python.org', 
  10.  
  11. 'http://www.python.org/about/', 
  12.  
  13. 'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html', 
  14.  
  15. 'http://www.python.org/doc/', 
  16.  
  17. 'http://www.python.org/download/', 
  18.  
  19. 'http://www.python.org/getit/', 
  20.  
  21. 'http://www.python.org/community/', 
  22.  
  23. 'https://wiki.python.org/moin/', 
  24.  
  25. 'http://planet.python.org/', 
  26.  
  27. 'https://wiki.python.org/moin/LocalUserGroups', 
  28.  
  29. 'http://www.python.org/psf/', 
  30.  
  31. 'http://docs.python.org/devguide/', 
  32.  
  33. 'http://www.python.org/community/awards/' 
  34.  
  35. # etc.. 
  36.  
  37. ] 
  38.  
  39.   
  40.  
  41. # Make the Pool of workers 
  42.  
  43. pool = ThreadPool(4) 
  44.  
  45. # Open the urls in their own threads 
  46.  
  47. # and return the results 
  48.  
  49. results = pool.map(urllib2.urlopen, urls) 
  50.  
  51. #close the pool and wait for the work to finish 
  52.  
  53. pool.close() 
  54.  
  55. 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)建

  1. import os 
  2.  
  3. import PIL 
  4.  
  5.   
  6.  
  7. from multiprocessing import Pool 
  8.  
  9. from PIL import Image 
  10.  
  11.   
  12.  
  13. SIZE = (75,75) 
  14.  
  15. SAVE_DIRECTORY = 'thumbs' 
  16.  
  17.   
  18.  
  19. def get_image_paths(folder): 
  20.  
  21. return (os.path.join(folder, f) 
  22.  
  23. for f in os.listdir(folder) 
  24.  
  25. if 'jpeg' in f) 
  26.  
  27.   
  28.  
  29. def create_thumbnail(filename): 
  30.  
  31. im = Image.open(filename) 
  32.  
  33. im.thumbnail(SIZE, Image.ANTIALIAS) 
  34.  
  35. base, fname = os.path.split(filename) 
  36.  
  37. save_path = os.path.join(base, SAVE_DIRECTORY, fname) 
  38.  
  39. im.save(save_path) 
  40.  
  41.   
  42.  
  43. if __name__ == '__main__': 
  44.  
  45. folder = os.path.abspath( 
  46.  
  47. '11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840') 
  48.  
  49. os.mkdir(os.path.join(folder, SAVE_DIRECTORY)) 
  50.  
  51.   
  52.  
  53. images = get_image_paths(folder) 
  54.  
  55.   
  56.  
  57. for image in images: 
  58.  
  59.              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)的話：

  1. import os 
  2.  
  3. import PIL 
  4.  
  5.   
  6.  
  7. from multiprocessing import Pool 
  8.  
  9. from PIL import Image 
  10.  
  11.   
  12.  
  13. SIZE = (75,75) 
  14.  
  15. SAVE_DIRECTORY = 'thumbs' 
  16.  
  17.   
  18.  
  19. def get_image_paths(folder): 
  20.  
  21. return (os.path.join(folder, f) 
  22.  
  23. for f in os.listdir(folder) 
  24.  
  25. if 'jpeg' in f) 
  26.  
  27.   
  28.  
  29. def create_thumbnail(filename): 
  30.  
  31. im = Image.open(filename) 
  32.  
  33. im.thumbnail(SIZE, Image.ANTIALIAS) 
  34.  
  35. base, fname = os.path.split(filename) 
  36.  
  37. save_path = os.path.join(base, SAVE_DIRECTORY, fname) 
  38.  
  39. im.save(save_path) 
  40.  
  41.   
  42.  
  43. if __name__ == '__main__': 
  44.  
  45. folder = os.path.abspath( 
  46.  
  47. '11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840') 
  48.  
  49. os.mkdir(os.path.join(folder, SAVE_DIRECTORY)) 
  50.  
  51.   
  52.  
  53. images = get_image_paths(folder) 
  54.  
  55.   
  56.  
  57. pool = Pool() 
  58.  
  59.         pool.map(create_thumbnail,images) 
  60.  
  61.         pool.close() 
  62.  
  63.         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ù)。 

責(zé)任編輯：龐桂玉 來(lái)源： Python開(kāi)發(fā)者
Python代碼并行任務(wù)
相關(guān)推薦

2014-02-12 13:43:50

代碼并行任務(wù)

2022-04-09 09:11:33

Python

2020-08-19 10:30:25

代碼Python多線程

2016-12-02 08:53:18

Python一行代碼

2021-11-02 16:25:41

Python代碼技巧

2020-09-28 12:34:38

Python代碼開(kāi)發(fā)

2020-08-12 14:54:00

Python代碼開(kāi)發(fā)

2017-04-05 11:10:23

Javascript代碼前端

2013-11-29 13:14:30

代碼網(wǎng)頁(yè)設(shè)計(jì)

2022-09-28 10:12:50

Python代碼可視化

2020-01-10 22:56:56

Python圖像處理Linux

2024-05-31 13:14:05

2020-09-09 16:00:22

Linux進(jìn)程

2021-08-31 09:49:37

CPU執(zhí)行語(yǔ)言

2021-04-19 10:38:06

代碼開(kāi)發(fā)工具

2023-11-10 09:41:44

Python代碼

2022-02-23 14:37:48

代碼Pythonbug

2024-11-08 17:22:22

2023-01-12 08:07:03

Python代碼版權(quán)

2020-02-14 12:26:55

Python愛(ài)心情人節(jié)
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號(hào)