1. 程式人生 > >用map函式來完成Python並行任務的簡單示例

用map函式來完成Python並行任務的簡單示例

眾所周知,Python的並行處理能力很不理想。我認為如果不考慮執行緒和GIL的標準引數(它們大多是合法的),其原因不是因為技術不到位,而是我們的使用方法不恰當。大多數關於Python執行緒和多程序的教材雖然都很出色,但是內容繁瑣冗長。它們的確在開篇鋪陳了許多有用資訊,但往往都不會涉及真正能提高日常工作的部分。

經典例子

DDG上以“Python threading tutorial (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。

我現在並不想說明使用Producer / Consume來解決執行緒/多程序的方法是錯誤的——因為它肯定正確,而且在很多情況下它是最佳方法。但我不認為這是平時寫程式碼的最佳選擇。

它的問題所在(個人觀點)

首先,你需要建立一個樣板式的鋪墊類。然後,你再建立一個佇列,通過其傳遞物件和監管佇列的兩端來完成任務。(如果你想實現資料的交換或儲存,通常還涉及另一個佇列的參與)。

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語句。而這些還僅僅只是繁瑣的開始!

我們目前為止都完成了什麼?基本上什麼都沒有。上面的程式碼幾乎一直都只是在進行傳遞。這是很基礎的方法,很容易出錯(該死,我剛才忘了在佇列物件上還需要呼叫task_done()方法(但是我懶得修改了)),價效比很低。還好,我們還有更好的方法。

介紹:Map

Map是一個很棒的小功能,同時它也是Python並行程式碼快速執行的關鍵。給不熟悉的人講解一下吧,map是從函式語言Lisp來的。map函式能夠按序映射出另一個函式。例如

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的方法。

好了,文章結束了。一行完成並行任務。

相關推薦

map函式完成Python並行任務簡單示例

眾所周知,Python的並行處理能力很不理想。我認為如果不考慮執行緒和GIL的標準引數(它們大多是合法的),其原因不是因為技術不到位,而是我們的使用方法不恰當。大多數關於Python執行緒和多程序的教材雖然都很出色,但是內容繁瑣冗長。它們的確在開篇鋪陳了許多

pythonmap()函式和多執行緒threading、Thread採集注意事項

import re import requests from threading import Thread def getUrl(): url = 'http://www.xiaohuar

python 定時任務排程】 APScheduler排程框架完成python指令碼排程

# -*- coding: utf-8 -*- def main_job1(): print('This job is run every 20 seconds.') # -*- co

廖雪峰python教程之map函式實現大小寫轉換

題目:利用map()函式,把使用者輸入的不規範的英文名字,變為首字母大寫,其他小寫的規範名字。輸入:['adam', 'LISA', 'barT'],輸出:['Adam', 'Lisa', 'Bart'] def normalize(name): return n

交換函式理解指標

寫一個函式,交換a和b的值 #include<stdio.h> void swap1(int a,int b) {  int tmp=a;  a=b;  b=tmp; } int main() {   int a=10;   int b=20;   print

QueueUserAPC() 函式強制執行緒退出等待狀態

         當系統建立一個執行緒的時候,會同時建立一個與執行緒相關的佇列。這個佇列被成為非同步過程呼叫(APC)佇列。同時我們也知道,windows提供了6個函式可以等待並將執行緒設定為可提醒狀態--------- SleepEx(),WaitForSingleObje

請編寫相關函式完成對一段英文字串進行字母統計功能,出現最多那個字母的出現次數,不區分大小寫。

思路:1、對字串進行過濾,把英文字母過濾出來。2、將所有英文字母統一轉成小寫(大寫)。3、將待求的字串與26個字母進行比較並統計個數4、統計出現最多的字母。原始碼:#include <stdio.h>#include <stdlib.h>#includ

C++11實現的執行緒池,可以使用類成員函式新增執行緒任務

功能:         1:可以使用類成員函式/全域性函式單獨的建立一個執行緒,可以帶多個引數。        2:可以使用類成員函式/全域性函式單獨的為執行緒池新增一個任務,可以帶多個引數。        3:執行緒池的執行緒數量可手動擴充套件,稍作修改可以修改為自動擴充,

JavaSE8基礎 鏈式編程 調方法返回對象再調方法 簡單示例

ima jdk win dem 鏈式 教程 hello world log os :windows7 x64 jdk:jdk-8u131-windows-x64 ide:Eclipse Oxygen Release (4.7.0) 代碼:

Python 裝飾器簡單示例

裝飾器簡單裝飾器示例: def servlet(func): print("into servlet")#1 print(servlet)#2 def foo(): print("into foo")#7 print(func)#8,真正的bar函數

JavaWeb三種常用的定時任務簡單示例

(1)、程式碼如下: package com.sundy.task; import java.util.Timer; import java.util.TimerTask; import javax.servlet.ServletContextEvent; import javax.servle

在c#多執行緒使用IOCP(完成埠)的簡單示例

在c#使用IOCP(完成埠)的簡單示例 上次給大家發了利用winsock原生的api來做一個同步的socket伺服器的例子,大致上只是貼了一些程式碼,相信大家這麼冰雪聰明,已經研究的差不多了。因為winsock的api使用在msdn或者google上都能很方便的查到,所以我

python】詳解map函式的用法之函式並行作用解析

Python函式程式設計中的map(func, seq1[, seq2,…]) 函式是將func作用於seq中的每一個元素,其中seq須是可迭代物件,並將所有的呼叫的結果作為一個list返回。如果func為None,作用同zip()。 本文參考自:Python中map()函式淺析一文,感謝精彩分享。 下面

mongodump以及mongorestore完成mongo的遷移任務

首先貼上官網說明:       詳細請見:https://docs.mongodb.com/manual/   在實際操作中,一般只需用到 mongodump -h ip:port -d dbName -o path   mongor

【程序5】 題目:利用條件運算符的嵌套完成此題:學習成績>=90分的同學A表示,60-89分之間的B表示,60分以下的C表示。 1.程序分析:(a>b)?a:b這是條件運算符的基本例子。

window code 例子 prompt 利用 學習 amp text span if…else語句相對比較多,但是容易理解 1 var scroe = window.prompt("請輸入1-100之間的數") 2 scroe = parseInt

【在線】使用在線軟件完成任務

作圖 roc 時間 itl run 文檔 瀏覽器 實時 -- 在線畫畫 ProcessOn   百度腦圖 在線辦公軟件 user.me 在線寫作 ShowDoc   看雲 在線獲取編程知識 菜鳥教程   MobDevGroup    在線設計APP原型圖 墨刀【在線】使用在

人臉檢測? 對Python來說太簡單, 調dlib包就可以完成

cool gpu sub lib 描述 依賴 ruby face 使用 “Dlib 是一個現代化的 C ++ 工具包,包含用於創建復雜軟件的機器學習算法和工具 ” 。它使您能夠直接在 Python 中運行許多任務,其中一個例子就是人臉檢測。 安裝 dlib 並不像只做一個

除錯經驗——條條大路通羅馬(能多種方法完成一件任務的重要性)

使用CitrixReciver訪問公司內網,發現密碼今天就要過期了(好險!)。 於是,使用首頁的“Change Password”連結,打不開。 無奈,在Webapp中搜索“change password”,找到了一個change password的新連結,但問題是,不能成功修改。

python 使用生成器 完成 監聽檔案輸入的例子

def tail(filename):#函式 f = open(filename,encoding='utf-8') while True: line = f.readline() if line.strip(): yield line.strip() g = tail('file')#引用 設定檔

學渣學pythonmap函式

map()函式是Python內建的高階函式,它接收一個函式f和一個list,並把函式f作用在list的每個元素上。從而得到一個f處理過的新的list返回。下面舉個栗子: 1. 例1 list [1, 2, 3, 4, 5, 6, 7] 我們要得到list的每個元素都平方後的,新的li