python全棧開發基礎【第二十六篇】(concurrent.futures模塊、協程、Greenlet、Gevent)
註意
1.不能無限的開進程,不能無限的開線程
最常用的就是開進程池,開線程池。其中回調函數非常重要
回調函數其實可以作為一種編程思想,誰好了誰就去掉
2.只要你用並發,就會有鎖的問題,但是你不能一直去自己加鎖吧
那麽我們就用QUEUE,這樣還解決了自動加鎖的問題
由Queue延伸出的一個點也非常重要的概念。以後寫程序也會用到
這個思想。就是生產者與消費者問題
一、Python標準模塊--concurrent.futures(並發未來)
concurent.future模塊需要了解的
1.concurent.future模塊是用來創建並行的任務,提供了更高級別的接口, 為了異步執行調用 2.concurent.future這個模塊用起來非常方便,它的接口也封裝的非常簡單3.concurent.future模塊既可以實現進程池,也可以實現線程池 4.模塊導入進程池和線程池 from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor 還可以導入一個Executor,但是你別這樣導,這個類是一個抽象類 抽象類的目的是規範他的子類必須有某種方法(並且抽象類的方法必須實現),但是抽象類不能被實例化 5. p = ProcessPoolExecutor(max_works)對於進程池如果不寫max_works:默認的是cpu的數目,默認是4個 p = ThreadPoolExecutor(max_works)對於線程池如果不寫max_works:默認的是cpu的數目*5 6.如果是進程池,得到的結果如果是一個對象。我們得用一個.get()方法得到結果,但是現在用了concurent.future模塊,我們可以用obj.result方法 p.submit(task,i)#相當於apply_async異步方法 p.shutdown() #默認有個參數wite=True (相當於close和join)
二、線程池
進程池:就是在一個進程內控制一定個數的線程
基於concurent.future模塊的進程池和線程池 (他們的同步執行和異步執行是一樣的)
# 基於concurrent.futures模塊的進程池 # 1.同步執行-------------- from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor import os,time,random def task(n): print(‘[%s] is running‘%os.getpid()) time.sleep(random.randint(1,3)) #I/O密集型的,,一般用線程,用了進程耗時長 return n**2 if __name__ == ‘__main__‘: start = time.time() p = ProcessPoolExecutor() for i in range(10): #現在是開了10個任務, 那麽如果是上百個任務呢,就不能無線的開進程,那麽就得考慮控制 # 線程數了,那麽就得考慮到池了 obj = p.submit(task,i).result() #相當於apply同步方法 p.shutdown() #相當於close和join方法 print(‘=‘*30) print(time.time() - start) #17.36499309539795 # 2.異步執行----------- # from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor # import os,time,random # def task(n): # print(‘[%s] is running‘%os.getpid()) # time.sleep(random.randint(1,3)) #I/O密集型的,,一般用線程,用了進程耗時長 # return n**2 # if __name__ == ‘__main__‘: # start = time.time() # p = ProcessPoolExecutor() # l = [] # for i in range(10): #現在是開了10個任務, 那麽如果是上百個任務呢,就不能無線的開進程,那麽就得考慮控制 # # 線程數了,那麽就得考慮到池了 # obj = p.submit(task,i) #相當於apply_async()異步方法 # l.append(obj) # p.shutdown() #相當於close和join方法 # print(‘=‘*30) # print([obj.result() for obj in l]) # print(time.time() - start) #5.362306594848633
#基於concurrent.futures模塊的線程池 from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor from threading import currentThread import os,time,random def task(n): print(‘%s:%s is running‘%(currentThread().getName(),os.getpid())) #看到的pid都是一樣的,因為線程是共享了一個進程 time.sleep(random.randint(1,3)) #I/O密集型的,,一般用線程,用了進程耗時長 return n**2 if __name__ == ‘__main__‘: start = time.time() p = ThreadPoolExecutor() #線程池 #如果不給定值,默認cup*5 l = [] for i in range(10): #10個任務 # 線程池效率高了 obj = p.submit(task,i) #相當於apply_async異步方法 l.append(obj) p.shutdown() #默認有個參數wite=True (相當於close和join) print(‘=‘*30) print([obj.result() for obj in l]) print(time.time() - start) #3.001171827316284
應用線程池(下載網頁並解析)
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor import requests import time,os def get_page(url): print(‘<%s> is getting [%s]‘%(os.getpid(),url)) response = requests.get(url) if response.status_code==200: #200代表狀態:下載成功了 return {‘url‘:url,‘text‘:response.text} def parse_page(res): res = res.result() print(‘<%s> is getting [%s]‘%(os.getpid(),res[‘url‘])) with open(‘db.txt‘,‘a‘) as f: parse_res = ‘url:%s size:%s\n‘%(res[‘url‘],len(res[‘text‘])) f.write(parse_res) if __name__ == ‘__main__‘: # p = ThreadPoolExecutor() p = ProcessPoolExecutor() l = [ ‘http://www.baidu.com‘, ‘http://www.baidu.com‘, ‘http://www.baidu.com‘, ‘http://www.baidu.com‘, ] for url in l: res = p.submit(get_page,url).add_done_callback(parse_page) #這裏的回調函數拿到的是一個對象。得 # 先把返回的res得到一個結果。即在前面加上一個res.result() #誰好了誰去掉回調函數 # 回調函數也是一種編程思想。不僅開線程池用,開線程池也用 p.shutdown() #相當於進程池裏的close和join print(‘主‘,os.getpid())
map函數的應用
# map函數舉例 obj= map(lambda x:x**2 ,range(10)) print(list(obj)) #運行結果[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
可以和上面的開進程池/線程池的對比著看,就能發現map函數的強大了
#map函數應用 # 我們的那個p.submit(task,i)和map函數的原理類似。我們就 # 可以用map函數去代替。更減縮了代碼 from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor import os,time,random def task(n): print(‘[%s] is running‘%os.getpid()) time.sleep(random.randint(1,3)) #I/O密集型的,,一般用線程,用了進程耗時長 return n**2 if __name__ == ‘__main__‘: p = ProcessPoolExecutor() obj = p.map(task,range(10)) p.shutdown() #相當於close和join方法 print(‘=‘*30) print(obj) #返回的是一個叠代器 print(list(obj))
三、協程介紹
協程:單進程下實現並發(提高效率),協程實質上是對單線程在時間上的充分利用,尤其是在IO密集型任務是,遇到IO切換到別的任務,返回結果由回到函數處理
說到協成,我們先說一下協程聯想到的知識點
切換關鍵的一點是:保存狀態(從原來停留的地方繼續切) return:只能執行一次,結束函數的標誌 yield:函數中但凡有yield,這個函數的執行結果就變成了一個生成器, 生成器本質就是一個叠代器,那麽叠代器怎麽用呢?用一個next()方法 1.yield語句的形式:yield 1 yield功能1:可以用來返回值,可以返回多次值 yield功能2:可以吧函數暫停住,保存原來的狀態 2.yield表達式的形式:x = yieldsend可以吧一個函數的結果傳給另一個函數,以此實現單線程內程序之間的切換 send()要想用就得先next()一下 但是要用send至少要用兩個yield
yield復習
# yield功能(可以吧函數暫停住,保存原來的狀態)-------------- def f1(): print(‘first‘) yield 1 print(‘second‘) yield 2 print(‘third‘) yield 3 # print(f1()) #加了yield返回的是一個生成器 g = f1() print(next(g)) #當遇見了yield的時候就返回一個值,而且保存原來的狀態 print(next(g)) #當遇見了yield的時候就返回一個值 print(next(g)) #當遇見了yield的時候就返回一個值
# yield表達式(對於表達式的yield)-------------------- import time def wrapper(func): def inner(*args,**kwargs): ret =func(*args,**kwargs) next(ret) return ret return inner @wrapper def consumer(): while True: x= yield print(x) def producter(target): ‘‘‘生產者造值‘‘‘ # next(g) #相當於g.send(None) for i in range(10): time.sleep(0.5) target.send(i)#要用send就得用兩個yield producter(consumer())
引子
本節主題是實現單線程下的並發,即只在一個主線程,並且很明顯的是,可利用的cpu只有一個情況下實現並發,
為此我們需要先回顧下並發的本質:切換+保存狀態
cpu正在運行一個任務,會在兩種情況下切走去執行其他的任務(切換由操作系統強制控制),
一種情況是該任務發生了阻塞,另外一種情況是該任務計算的時間過長
其中第二種情況並不能提升效率,只是為了讓cpu能夠雨露均沾,實現看起來大家都被執行的效果,如果多個程序都是純計算任務,這種切換反而會降低效率。為此我們可以基於yield來驗證。yield本身就是一種在單線程下可以保存任務運行狀態的方法,我們來簡單復習一下:
yiled可以保存狀態,
yield
的狀態保存與操作系統的保存線程狀態很像,但是
yield
是代碼級別控制的,更輕量級
send可以把一個函數的結果傳給另外一個函數,以此實現單線程內程序之間的切換
單純的切反而會影響效率
#串行執行 import time def consumer(res): ‘‘‘任務1:接收數據,處理數據‘‘‘ pass def producer(): ‘‘‘任務2:生產數據‘‘‘ res=[] for i in range(10000000): res.append(i) return res start=time.time() #串行執行 res=producer() consumer(res) stop=time.time() print(stop-start) #1.5536692142486572
# 基於yield並發執行 import time def wrapper(func): def inner(*args,**kwargs): ret =func(*args,**kwargs) next(ret) return ret return inner @wrapper def consumer(): while True: x= yield print(x) def producter(target): ‘‘‘生產者造值‘‘‘ # next(g) #相當於g.send(None) for i in range(10): time.sleep(0.5) target.send(i)#要用send就得用兩個yield producter(consumer())
對於單線程下,我們不可避免程序中出現io操作,但如果我們能在自己的程序中(即用戶程序級別,而非操作系統級別)控制單線程下多個任務能遇到io就切換,這樣就保證了該線程能夠最大限度地處於就緒態,即隨時都可以被cpu執行的狀態,相當於我們在用戶程序級別將自己的io操作最大限度地隱藏起來,對於操作系統來說:這哥們(該線程)好像是一直處於計算過程的,io比較少。
協程的本質就是在單線程下,由用戶自己控制一個任務遇到io阻塞了就切換另外一個任務去執行,以此來提升效率。
因此我們需要找尋一種可以同時滿足以下條件的解決方案:
1. 可以控制多個任務之間的切換,切換之前將任務的狀態保存下來(重新運行時,可以基於暫停的位置繼續)
2. 作為1的補充:可以檢測io操作,在遇到io操作的情況下才發生切換
四、Greenlet
Greenlet模塊和yield沒有什麽區別,就只是單純的切,跟效率無關。
只不過比yield更好一點,切的時候方便一點。但是仍然沒有解決效率
Greenlet可以讓你在多個任務之間來回的切
# greenlet示例 from greenlet import greenlet import time def eat(name): print(‘%s eat 1‘ %name) time.sleep(10) #當遇到IO的時候它也沒有切,這就得用gevent了 g2.switch(‘egon‘) print(‘%s eat 2‘ %name) g2.switch() def play(name): print(‘%s play 1‘ %name) g1.switch() print(‘%s play 2‘ %name) g1=greenlet(eat) g2=greenlet(play) g1.switch(‘egon‘)#可以在第一次switch時傳入參數,以後都不需要
所以上面的方法都不可行,那麽這就用到了Gevert ,也就是協程。就解決了單線程實現並發的問題,還提升了效率
五、Gevent介紹
Gevent 是一個第三方庫,可以輕松通過gevent實現並發同步或異步編程,在gevent中用到的主要模式是Greenlet,
它是以C擴展模塊形式接入Python的輕量級協程。 Greenlet全部運行在主程序操作系統進程的內部,但它們被協作式地調度。
#用法 g1=gevent.spawn(func,1,,2,3,x=4,y=5)創建一個協程對象g1,spawn括號內第一個參數是函數名,如eat,後面可以有多個參數,可以是位置實參或關鍵字實參,都是傳給函數eat的 g2=gevent.spawn(func2) g1.join() #等待g1結束 g2.join() #等待g2結束 #或者上述兩步合作一步:gevent.joinall([g1,g2]) g1.value#拿到func1的返回值
舉例;
# gevent的一些方法(重要) from gevent import monkey;monkey.patch_all() import gevent import time def eat(name): print(‘%s eat 1‘ %name) time.sleep(2) #我們用等待的時間模擬IO阻塞 ‘‘‘ 在gevent模塊裏面要用gevent.sleep(2)表示等待的時間 然而我們經常用time.sleep()用習慣了,那麽有些人就想著 可以用time.sleep(),那麽也不是不可以。要想用,就得在 最上面導入from gevent import monkey;monkey.patch_all()這句話 如果不導入直接用time.sleep(),就實現不了單線程並發的效果了 ‘‘‘ # gevent.sleep(2) print(‘%s eat 2‘ %name) return ‘eat‘ def play(name): print(‘%s play 1‘ %name) time.sleep(3) # gevent.sleep(3) print(‘%s play 2‘ %name) return ‘paly‘ #當有返回值的時候,gevent模塊也提供了返回結果的操作 start = time.time() g1 = gevent.spawn(eat,‘egon‘) #執行任務 g2 = gevent.spawn(play,‘egon‘) #g1和g2的參數可以不一樣 # g1.join() #等待g1 # g2.join() #等待g2 #上面等待的兩句也可以這樣寫 gevent.joinall([g1,g2]) print(‘主‘,time.time()-start) #3.001171588897705 print(g1.value) print(g2.value)
需要說明的是:
gevent.sleep(2)模擬的是gevent可以識別的io阻塞,
而time.sleep(2)或其他的阻塞,gevent是不能直接識別的需要用下面一行代碼,打補丁,就可以識別了
from gevent import monkey;monkey.patch_all()必須放到被打補丁者的前面,如time,socket模塊之前
或者我們幹脆記憶成:要用gevent,需要將from gevent import monkey;monkey.patch_all()放到文件的開頭
六、Gevent之同步於異步
from gevent import spawn,joinall,monkey;monkey.patch_all() import time def task(pid): """ Some non-deterministic task """ time.sleep(0.5) print(‘Task %s done‘ % pid) def synchronous(): for i in range(10): task(i) def asynchronous(): g_l=[spawn(task,i) for i in range(10)] joinall(g_l) if __name__ == ‘__main__‘: print(‘Synchronous:‘) synchronous() print(‘Asynchronous:‘) asynchronous() #上面程序的重要部分是將task函數封裝到Greenlet內部線程的gevent.spawn。 初始化的greenlet列表存放在數組threads中,此數組被傳給gevent.joinall 函數,後者阻塞當前流程,並執行所有給定的greenlet。執行流程只會在 所有greenlet執行完後才會繼續向下走。
七、Gevent之應用舉例一
#協程應用爬蟲 from gevent import monkey;monkey.patch_all() #打補丁 import gevent import requests import time def get_page(url): print(‘get :%s‘%url) response = requests.get(url) if response.status_code==200: #下載成功的狀態 print(‘%d bytes received from:%s‘%(len(response.text),url)) start=time.time() gevent.joinall([ gevent.spawn(get_page,‘http://www.baidu.com‘), gevent.spawn(get_page, ‘https://www.yahoo.com/‘), gevent.spawn(get_page, ‘https://github.com/‘), ]) stop = time.time() print(‘run time is %s‘ %(stop-start))
#協程應用爬蟲加了回調函數的 from gevent import joinall,spawn,monkey;monkey.patch_all() import requests from threading import current_thread def parse_page(res): print(‘%s PARSE %s‘ %(current_thread().getName(),len(res))) def get_page(url,callback=parse_page): print(‘%s GET %s‘ %(current_thread().getName(),url)) response=requests.get(url) if response.status_code == 200: callback(response.text) if __name__ == ‘__main__‘: urls=[ ‘https://www.baidu.com‘, ‘https://www.taobao.com‘, ‘https://www.openstack.org‘, ] tasks=[] for url in urls: tasks.append(spawn(get_page,url)) joinall(tasks)
八、Gevent之應用舉例二
也可以利用協程實現並發
# 服務端利用協程 #!usr/bin/env python # -*- coding:utf-8 -*- from gevent import monkey;monkey.patch_all() import gevent from socket import * print(‘start running...‘) def talk(conn,addr): while True: data = conn.recv(1024) print(‘%s:%s %s‘%(addr[0],addr[1],data)) conn.send(data.upper()) conn.close() def server(ip,duankou): server = socket(AF_INET, SOCK_STREAM) server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) server.bind((ip,duankou)) server.listen(5) while True: conn,addr = server.accept() #等待鏈接 gevent.spawn(talk,conn,addr) #異步執行 (p =Process(target=talk,args=(coon,addr)) # p.start())相當於開進程裏的這兩句 server.close() if __name__ == ‘__main__‘: server(‘127.0.0.1‘,8081)
#客戶端開了100個進程 #!usr/bin/env python # -*- coding:utf-8 -*- from multiprocessing import Process from gevent import monkey;monkey.patch_all() from socket import * def client(ip,duankou): client = socket(AF_INET, SOCK_STREAM) client.connect((ip,duankou)) while True: client.send(‘hello‘.encode(‘utf-8‘)) data = client.recv(1024) print(data.decode(‘utf-8‘)) if __name__ == ‘__main__‘: for i in range(100): p = Process(target=client,args=((‘127.0.0.1‘,8081))) p.start()
python全棧開發基礎【第二十六篇】(concurrent.futures模塊、協程、Greenlet、Gevent)