深入了解Python并發(fā)編程
并發(fā)方式
線程（[Thread]）
多線程幾乎是每一個程序猿在使用每一種語言時都會首先想到用于解決并發(fā)的工具（JS程序員請回避），使用多線程可以有效的利用CPU資源（Python例外）。然而多線程所帶來的程序的復雜度也不可避免，尤其是對競爭資源的同步問題。
然而在python中由于使用了全局解釋鎖（GIL）的原因，代碼并不能同時在多核上并發(fā)的運行，也就是說，Python的多線程不能并發(fā)，很多人會發(fā)現(xiàn)使用多線程來改進自己的Python代碼后，程序的運行效率卻下降了，這是多么蛋疼的一件事呀！實際上使用多線程的編程模型是很困難的，程序員很容易犯錯，這并不是程序員的錯誤，因為并行思維是反人類的，我們大多數(shù)人的思維是串行（精神分裂不討論），而且馮諾依曼設計的計算機架構也是以順序執(zhí)行為基礎的。所以如果你總是不能把你的多線程程序搞定，恭喜你，你是個思維正常的程序猿：）
Python提供兩組線程的接口，一組是thread模塊，提供基礎的，低等級（Low Level）接口，使用Function作為線程的運行體。還有一組是threading模塊，提供更容易使用的基于對象的接口（類似于Java），可以繼承Thread對象來實現(xiàn)線程，還提供了其它一些線程相關的對象，例如Timer，Lock
使用thread模塊的例子
import thread
def worker():
"""thread worker function"""
print 'Worker'
thread.start_new_thread(worker)使用threading模塊的例子
import threading
def worker():
"""thread worker function"""
print 'Worker'
t = threading.Thread(target=worker)
t.start()或者Java Style
import threading
class worker(threading.Thread):
def __init__(self):
pass
def run():
"""thread worker function"""
print 'Worker'
t = worker()
t.start()進程 （Process）
由于前文提到的全局解釋鎖的問題，Python下比較好的并行方式是使用多進程，這樣可以非常有效的使用CPU資源，并實現(xiàn)真正意義上的并發(fā)。當然，進程的開銷比線程要大，也就是說如果你要創(chuàng)建數(shù)量驚人的并發(fā)進程的話，需要考慮一下你的機器是不是有一顆強大的心。
Python的mutliprocess模塊和threading具有類似的接口。
from multiprocessing import Process
def worker():
"""thread worker function"""
print 'Worker'
p = Process(target=worker)
p.start()
p.join()由于線程共享相同的地址空間和內存，所以線程之間的通信是非常容易的，然而進程之間的通信就要復雜一些了。常見的進程間通信有，管道，消息隊列，Socket接口（TCP/IP）等等。
Python的mutliprocess模塊提供了封裝好的管道和隊列，可以方便的在進程間傳遞消息。
Python進程間的同步使用鎖，這一點喝線程是一樣的。
另外，Python還提供了進程池Pool對象，可以方便的管理和控制線程。
遠程分布式主機 （Distributed Node）
隨著大數(shù)據(jù)時代的到臨，摩爾定理在單機上似乎已經(jīng)失去了效果，數(shù)據(jù)的計算和處理需要分布式的計算機網(wǎng)絡來運行，程序并行的運行在多個主機節(jié)點上，已經(jīng)是現(xiàn)在的軟件架構所必需考慮的問題。
遠程主機間的進程間通信有幾種常見的方式
- TCP／IP
TCP／IP是所有遠程通信的基礎，然而API比較低級別，使用起來比較繁瑣，所以一般不會考慮
- 遠程方法調用 Remote Function Call
[RPC]
- 遠程對象 Remote Object
遠程對象是更高級別的封裝，程序可以想操作本地對象一樣去操作一個遠程對象在本地的代理。遠程對象最廣為使用的規(guī)范CORBA，CORBA最大的好處是可以在不同語言和平臺中進行通信。當讓不用的語言和平臺還有一些各自的遠程對象實現(xiàn)，例如Java的RMI，MS的DCOM
Python的開源實現(xiàn)，有許多對遠程對象的支持
- Dopy]
- Fnorb （CORBA）
- ICE
- omniORB （CORBA）
- Pyro
- YAMI
- 消息隊列 Message Queue
比起RPC或者遠程對象，消息是一種更為靈活的通信手段，常見的支持Python接口的消息機制有
- RabbitMQ
- ZeroMQ
- Kafka
- AWS SQS ＋ BOTO
在遠程主機上執(zhí)行并發(fā)和本地的多進程并沒有非常大的差異，都需要解決進程間通信的問題。當然對遠程進程的管理和協(xié)調比起本地要復雜。
Python下有許多開源的框架來支持分布式的并發(fā)，提供有效的管理手段包括：
- Celery
Celery是一個非常成熟的Python分布式框架，可以在分布式的系統(tǒng)中，異步的執(zhí)行任務，并提供有效的管理和調度功能。
- SCOOP
SCOOP （Scalable COncurrent Operations in Python）提供簡單易用的分布式調用接口，使用Future接口來進行并發(fā)。
- Dispy
相比起Celery和SCOOP，Dispy提供更為輕量級的分布式并行服務
- PP
PP （Parallel Python）是另外一個輕量級的Python并行服務
- Asyncoro
Asyncoro是另一個利用Generator實現(xiàn)分布式并發(fā)的Python框架，
當然還有許多其它的系統(tǒng)，我沒有一一列出
另外，許多的分布式系統(tǒng)多提供了對Python接口的支持，例如Spark
偽線程 （Pseudo－Thread）
還有一種并發(fā)手段并不常見，我們可以稱之為偽線程，就是看上去像是線程，使用的接口類似線程接口，但是實際使用非線程的方式，對應的線程開銷也不存的。
- greenlet
greenlet提供輕量級的coroutines來支持進程內的并發(fā)。
greenlet是Stackless的一個副產(chǎn)品，使用tasklet來支持一中被稱之為微線程（mirco－thread）的技術，這里是一個使用greenlet的偽線程的例子
from greenlet import greenlet
def test1():
print 12
gr2.switch()
print 34
def test2():
print 56
gr1.switch()
print 78
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()運行以上程序得到如下結果：
12
56
34
偽線程gr1 switch會打印12，然后調用gr2 switch得到56，然后switch回到gr1，打印34，然后偽線程gr1結束，程序退出，所以78永遠不會被打印。通過這個例子我們可以看出，使用偽線程，我們可以有效的控制程序的執(zhí)行流程，但是偽線程并不存在真正意義上的并發(fā)。
eventlet，gevent和concurence都是基于greenlet提供并發(fā)的。
eventlet是一個提供網(wǎng)絡調用并發(fā)的Python庫，使用者可以以非阻塞的方式來調用阻塞的IO操作。
import eventlet
from eventlet.green import urllib2
urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org']
def fetch(url):
return urllib2.urlopen(url).read()
pool = eventlet.GreenPool()
for body in pool.imap(fetch, urls):
print("got body", len(body))執(zhí)行結果如下
('got body', 17629)
('got body', 1270)
('got body', 46949)
eventlet為了支持generator的操作對urllib2做了修改，接口和urllib2是一致的。這里的GreenPool和Python的Pool接口一致。
- gevent
gevent和eventlet類似，
import gevent from gevent import socket urls = ['www.google.com', 'www.example.com', 'www.python.org'] jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls] gevent.joinall(jobs, timeout=2) print [job.value for job in jobs]
執(zhí)行結果如下：
['206.169.145.226', '93.184.216.34', '23.235.39.223']
concurence是另外一個利用greenlet提供網(wǎng)絡并發(fā)的開源庫，我沒有用過，大家可以自己嘗試一下。
實戰(zhàn)運用
通常需要用到并發(fā)的場合有兩種，一種是計算密集型，也就是說你的程序需要大量的CPU資源;另一種是IO密集型，程序可能有大量的讀寫操作，包括讀寫文件，收發(fā)網(wǎng)絡請求等等。
計算密集型
對應計算密集型的應用，我們選用著名的蒙特卡洛算法來計算PI值?；驹砣缦?/p>
蒙特卡洛算法利用統(tǒng)計學原理來模擬計算圓周率，在一個正方形中，一個隨機的點落在1/4圓的區(qū)域（紅色點）的概率與其面積成正比。也就該概率 p ＝ Pi ＊ R＊R ／4 ： R＊ R ， 其中R是正方形的邊長，圓的半徑。也就是說該概率是圓周率的1/4, 利用這個結論，只要我們模擬出點落在四分之一圓上的概率就可以知道圓周率了，為了得到這個概率，我們可以通過大量的實驗，也就是生成大量的點，看看這個點在哪個區(qū)域，然后統(tǒng)計出結果。
基本算法如下：
from math import hypot
from random import random
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))這里test方法做了n（tries）次試驗，返回落在四分之一圓中的點的個數(shù)。判斷方法是檢查該點到圓心的距離，如果小于R則是在圓上。
通過大量的并發(fā)，我們可以快速的運行多次試驗，試驗的次數(shù)越多，結果越接近真實的圓周率。
這里給出不同并發(fā)方法的程序代碼
- 非并發(fā)
我們先在單線程，但進程運行，看看性能如何
from math import hypot
from random import random
import eventlet
import time
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
result = map(test, [tries] * nbFutures)
ret = 4. * sum(result) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
print calcPi(3000,4000)- 多線程 thread
為了使用線程池，我們用multiprocessing的dummy包，它是對多線程的一個封裝。注意這里代碼雖然一個字的沒有提到線程，但它千真萬確是多線程。
通過測試我們開（jing）心（ya）的發(fā)現(xiàn)，果然不出所料，當線程池為1是，它的運行結果和沒有并發(fā)時一樣，當我們把線程池數(shù)字設置為5時，耗時幾乎是沒有并發(fā)的2倍，我的測試數(shù)據(jù)從5秒到9秒。所以對于計算密集型的任務，還是放棄多線程吧。
from multiprocessing.dummy import Pool
from math import hypot
from random import random
import time
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
p = Pool(1)
result = p.map(test, [tries] * nbFutures)
ret = 4. * sum(result) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
if __name__ == '__main__':
p = Pool()
print("pi = {}".format(calcPi(3000, 4000)))- 多進程 multiprocess
理論上對于計算密集型的任務，使用多進程并發(fā)比較合適，在以下的例子中，進程池的規(guī)模設置為5，修改進程池的大小可以看到對結果的影響，當進程池設置為1時，和多線程的結果所需的時間類似，因為這時候并不存在并發(fā)；當設置為2時，響應時間有了明顯的改進，是之前沒有并發(fā)的一半；然而繼續(xù)擴大進程池對性能影響并不大，甚至有所下降，也許我的Apple Air的CPU只有兩個核？
當心，如果你設置一個非常大的進程池，你會遇到 Resource temporarily unavailable的錯誤，系統(tǒng)并不能支持創(chuàng)建太多的進程，畢竟資源是有限的。
from multiprocessing import Pool
from math import hypot
from random import random
import time
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
p = Pool(5)
result = p.map(test, [tries] * nbFutures)
ret = 4. * sum(result) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
if __name__ == '__main__':
print("pi = {}".format(calcPi(3000, 4000)))- gevent （偽線程）
不論是gevent還是eventlet，因為不存在實際的并發(fā)，響應時間和沒有并發(fā)區(qū)別不大，這個和測試結果一致。
import gevent
from math import hypot
from random import random
import time
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
jobs = [gevent.spawn(test, t) for t in [tries] * nbFutures]
gevent.joinall(jobs, timeout=2)
ret = 4. * sum([job.value for job in jobs]) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
print calcPi(3000,4000)- eventlet （偽線程）
from math import hypot
from random import random
import eventlet
import time
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
pool = eventlet.GreenPool()
result = pool.imap(test, [tries] * nbFutures)
ret = 4. * sum(result) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
print calcPi(3000,4000)- SCOOP
SCOOP中的Future接口符合PEP-3148的定義，也就是在Python3中提供的Future接口。
在缺省的SCOOP配置環(huán)境下（單機，4個Worker），并發(fā)的性能有提高，但是不如兩個進程池配置的多進程。
from math import hypot
from random import random
from scoop import futures
import time
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
expr = futures.map(test, [tries] * nbFutures)
ret = 4. * sum(expr) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
if __name__ == "__main__":
print("pi = {}".format(calcPi(3000, 4000)))- Celery
任務代碼
from celery import Celery
from math import hypot
from random import random
app = Celery('tasks', backend='amqp', broker='amqp://guest@localhost//')
app.conf.CELERY_RESULT_BACKEND = 'db+sqlite:///results.sqlite'
@app.task
def test(tries):
return sum(hypot(random(), random()) < 1 for _ in range(tries))客戶端代碼
from celery import group
from tasks import test
import time
def calcPi(nbFutures, tries):
ts = time.time()
result = group(test.s(tries) for i in xrange(nbFutures))().get()
ret = 4. * sum(result) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
print calcPi(3000, 4000)使用Celery做并發(fā)的測試結果出乎意料（環(huán)境是單機，4frefork的并發(fā)，消息broker是rabbitMQ），是所有測試用例里最糟糕的，響應時間是沒有并發(fā)的5～6倍。這也許是因為控制協(xié)調的開銷太大。對于這樣的計算任務，Celery也許不是一個好的選擇。
- asyncoro
Asyncoro的測試結果和非并發(fā)保持一致。
import asyncoro
from math import hypot
from random import random
import time
def test(tries):
yield sum(hypot(random(), random()) < 1 for _ in range(tries))
def calcPi(nbFutures, tries):
ts = time.time()
coros = [ asyncoro.Coro(test,t) for t in [tries] * nbFutures]
ret = 4. * sum([job.value() for job in coros]) / float(nbFutures * tries)
span = time.time() - ts
print "time spend ", span
return ret
print calcPi(3000,4000)IO密集型
IO密集型的任務是另一種常見的用例，例如網(wǎng)絡WEB服務器就是一個例子，每秒鐘能處理多少個請求時WEB服務器的重要指標。
我們就以網(wǎng)頁讀取作為最簡單的例子
from math import hypot
import time
import urllib2
urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org']
def test(url):
return urllib2.urlopen(url).read()
def testIO(nbFutures):
ts = time.time()
map(test, urls * nbFutures)
span = time.time() - ts
print "time spend ", span
testIO(10)在不同并發(fā)庫下的代碼，由于比較類似，我就不一一列出。大家可以參考計算密集型中代碼做參考。
通過測試我們可以發(fā)現(xiàn)，對于IO密集型的任務，使用多線程，或者是多進程都可以有效的提高程序的效率，而使用偽線程性能提升非常顯著，eventlet比沒有并發(fā)的情況下，響應時間從9秒提高到0.03秒。同時eventlet／gevent提供了非阻塞的異步調用模式，非常方便。這里推薦使用線程或者偽線程，因為在響應時間類似的情況下，線程和偽線程消耗的資源更少。
總結
Python提供了不同的并發(fā)方式，對應于不同的場景，我們需要選擇不同的方式進行并發(fā)。選擇合適的方式，不但要對該方法的原理有所了解，還應該做一些測試和試驗，數(shù)據(jù)才是你做選擇的最好參考。
以上就是深入了解Python并發(fā)編程的詳細內容，更多關于Python并發(fā)編程的資料請關注腳本之家其它相關文章！
相關文章
tensorflow ckpt模型和pb模型獲取節(jié)點名稱,及ckpt轉pb模型實例
今天小編就為大家分享一篇tensorflow ckpt模型和pb模型獲取節(jié)點名稱,及ckpt轉pb模型實例，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
這篇文章主要介紹了python中序列的逆序方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-05-05
本文主要介紹了Python嵌套循環(huán)的使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧<BR>2023-02-02
人工智能學習pyTorch自建數(shù)據(jù)集及可視化結果實現(xiàn)過程
這篇文章主要為大家介紹了人工智能學習pyTorch自建數(shù)據(jù)集及可視化結果的實現(xiàn)過程，有需要的朋友可以借鑒參考下，希望能夠有所幫助2021-11-11
這篇文章主要介紹了利用機器學習回歸模型預測房價,解釋清晰,代碼詳細,是很不錯的機器學習實戰(zhàn)演練,對機器學習感興趣的朋友可以參考一下2021-04-04
這篇文章主要介紹了Python輸入若干整數(shù)求和方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-08-08
在Python中我們在項目中會用到各種庫,自帶的自然不必再說,這篇文章主要給大家介紹了關于Python如何導出導入所有依賴包的相關資料,需要的朋友可以參考下2021-06-06
這篇文章主要給大家介紹了關于如何用Python實現(xiàn)八數(shù)碼問題的相關資料,八數(shù)碼問題是一種經(jīng)典的搜索問題,它的目標是將一個亂序的八數(shù)碼序列變成一個有序的八數(shù)碼序列,通常使用 A* 算法來解決,需要的朋友可以參考下2023-10-10
這篇文章主要為大家詳細介紹了Python如何實現(xiàn)快速計算24點游戲并獲取表達式，文中的示例代碼講解詳細，感興趣的小伙伴可以了解一下2022-12-12
這篇文章主要介紹了python雙向鏈表原理與實現(xiàn)方法,結合實例形式分析了Python雙向鏈表的定義、以及節(jié)點的判斷、遍歷、添加、刪除等相關操作技巧,需要的朋友可以參考下2019-12-12