Python-併發程式設計(協程)
今天說說協程
一.引子
本節的主題是基於單執行緒來實現併發，即只用一個主執行緒（很明顯可利用的cpu只有一個）情況下實現併發，為此我們需要先回顧下併發的本質：切換+儲存狀態
cpu正在執行一個任務，會在兩種情況下切走去執行其他的任務（切換由作業系統強制控制），一種情況是該任務發生了阻塞，另外一種情況是該任務計算的時間過長或有一個優先順序更高的程式替代了它
協程本質上就是一個執行緒，以前執行緒任務的切換是由作業系統控制的，遇到I/O自動切換，現在我們用協程的目的就是較少作業系統切換的開銷（開關執行緒，建立暫存器、堆疊等，在他們之間進行切換等），在我們自己的程式裡面來控制任務的切換。
ps：在介紹程序理論時，提及程序的三種執行狀態，而執行緒才是執行單位，所以也可以將上圖理解為執行緒的三種狀態
一：其中第二種情況並不能提升效率，只是為了讓cpu能夠雨露均沾，實現看起來所有任務都被“同時”執行的效果，如果多個任務都是純計算的，這種切換反而會降低效率。為此我們可以基於yield來驗證。yield本身就是一種在單執行緒下可以儲存任務執行狀態的方法，我們來簡單複習一下：
#1 yiled可以儲存狀態，yield的狀態儲存與作業系統的儲存執行緒狀態很像，但是yield是程式碼級別控制的，更輕量級 #2 send可以把一個函式的結果傳給另外一個函式，以此實現單執行緒內程式之間的切換
通過yield實現任務切換+儲存現場
import time
def func1():
for i in range(11):
#yield
print('這是我第%s次列印啦' % i)
time.sleep(1)
def func2():
g = func1()
#next(g)
for k in range(10):
print('哈哈，我第%s次列印了' % k)
time.sleep(1)
#next(g)
#不寫yield，下面兩個任務是執行完func1裡面所有的程式才會執行func2裡面的程式，有了yield，我們實現了兩個任務的切換+儲存狀態
func1()
func2()
單純的切換反而會降低執行效率
#基於yield併發執行，多工之間來回切換，這就是個簡單的協程的體現，但是他能夠節省I/O時間嗎？不能
import time
def consumer():
'''任務1:接收資料,處理資料'''
while True:
x=yield
# time.sleep(1) #發現什麼？只是進行了切換，但是並沒有節省I/O時間
print('處理了資料：',x)
def producer():
'''任務2:生產資料'''
g=consumer()
next(g) #找到了consumer函式的yield位置
for i in range(3):
# for i in range(10000000):
g.send(i) #給yield傳值，然後再迴圈給下一個yield傳值，並且多了切換的程式，比直接序列執行還多了一些步驟，導致執行效率反而更低了。
print('傳送了資料：',i)
start=time.time()
#基於yield儲存狀態,實現兩個任務直接來回切換,即併發的效果
#PS:如果每個任務中都加上列印,那麼明顯地看到兩個任務的列印是你一次我一次,即併發執行的.
producer() #我在當前執行緒中只執行了這個函式，但是通過這個函式裡面的send切換了另外一個任務
stop=time.time()
# 序列執行的方式
# res=producer()
# consumer(res)
# stop=time.time()
print(stop-start)
二：第一種情況的切換。在任務一遇到io情況下，切到任務二去執行，這樣就可以利用任務一阻塞的時間完成任務二的計算，效率的提升就在於此。
yield不能檢測IO,實現遇到IO自動切換
import time
def func1():
while True:
print('func1')
yield
def func2():
g=func1()
for i in range(10000000):
i+1
next(g)
time.sleep(3)
print('func2')
start=time.time()
func2()
stop=time.time()
print(stop-start)
協程就是告訴Cpython直譯器，你不是nb嗎，不是搞了個GIL鎖嗎，那好，我就自己搞成一個執行緒讓你去執行，省去你切換執行緒的時間，我自己切換比你切換要快很多，避免了很多的開銷，對於單執行緒下，我們不可避免程式中出現io操作，但如果我們能在自己的程式中（即使用者程式級別，而非作業系統級別）控制單執行緒下的多個任務能在一個任務遇到io阻塞時就切換到另外一個任務去計算，這樣就保證了該執行緒能夠最大限度地處於就緒態，即隨時都可以被cpu執行的狀態，相當於我們在使用者程式級別將自己的io操作最大限度地隱藏起來，從而可以迷惑作業系統，讓其看到：該執行緒好像是一直在計算，io比較少，從而更多的將cpu的執行許可權分配給我們的執行緒。
協程的本質就是在單執行緒下，由使用者自己控制一個任務遇到io阻塞了就切換另外一個任務去執行，以此來提升效率。為了實現它，我們需要找尋一種可以同時滿足以下條件的解決方案：
#1. 可以控制多個任務之間的切換，切換之前將任務的狀態儲存下來，以便重新執行時，可以基於暫停的位置繼續執行。 #2. 作為1的補充：可以檢測io操作，在遇到io操作的情況下才發生切換
二.協程介紹
協程：是單執行緒下的併發，又稱微執行緒，纖程。英文名Coroutine。一句話說明什麼是執行緒：協程是一種使用者態的輕量級執行緒，即協程是由使用者程式自己控制排程的。、
需要強調的是：
#1. python的執行緒屬於核心級別的，即由作業系統控制排程（如單執行緒遇到io或執行時間過長就會被迫交出cpu執行許可權，切換其他執行緒執行） #2. 單執行緒內開啟協程，一旦遇到io，就會從應用程式級別（而非作業系統）控制切換，以此來提升效率（！！！非io操作的切換與效率無關）
對比作業系統控制執行緒的切換，使用者在單執行緒內控制協程的切換
優點如下：
#1. 協程的切換開銷更小，屬於程式級別的切換，作業系統完全感知不到，因而更加輕量級 #2. 單執行緒內就可以實現併發的效果，最大限度地利用cpu
缺點如下:
#1. 協程的本質是單執行緒下，無法利用多核，可以是一個程式開啟多個程序，每個程序內開啟多個執行緒，每個執行緒內開啟協程 #2. 協程指的是單個執行緒，因而一旦協程出現阻塞，將會阻塞整個執行緒
總結協程特點：
- 必須在只有一個單執行緒裡實現併發
- 修改共享資料不需加鎖
- 使用者程式裡自己儲存多個控制流的上下文棧
- 附加：一個協程遇到IO操作自動切換到其它協程（如何實現檢測IO，yield、greenlet都無法實現，就用到了gevent模組（select機制））
三.Greenlet
如果我們在單個執行緒內有20個任務，要想實現在多個任務之間切換，使用yield生成器的方式過於麻煩（需要先得到初始化一次的生成器，然後再呼叫send。。。非常麻煩），而使用greenlet模組可以非常簡單地實現這20個任務直接的切換
#安裝
pip3 install greenlet
#真正的協程模組就是使用greenlet完成的切換
from greenlet import greenlet
def eat(name):
print('%s eat 1' %name) #2
g2.switch('taibai') #3
print('%s eat 2' %name) #6
g2.switch() #7
def play(name):
print('%s play 1' %name) #4
g1.switch() #5
print('%s play 2' %name) #8
g1=greenlet(eat)
g2=greenlet(play)
g1.switch('taibai')#可以在第一次switch時傳入引數，以後都不需要 1
單純的切換（在沒有io的情況下或者沒有重複開闢記憶體空間的操作），反而會降低程式的執行速度
效率對比
#順序執行
import time
def f1():
res=1
for i in range(100000000):
res+=i
def f2():
res=1
for i in range(100000000):
res*=i
start=time.time()
f1()
f2()
stop=time.time()
print('run time is %s' %(stop-start)) #10.985628366470337
#切換
from greenlet import greenlet
import time
def f1():
res=1
for i in range(100000000):
res+=i
g2.switch()
def f2():
res=1
for i in range(100000000):
res*=i
g1.switch()
start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('run time is %s' %(stop-start)) # 52.763017892837524
greenlet只是提供了一種比generator更加便捷的切換方式，當切到一個任務執行時如果遇到io，那就原地阻塞，仍然是沒有解決遇到IO自動切換來提升效率的問題。
上面這個圖，是協程真正的意義，雖然沒有規避固有的I/O時間，但是我們使用這個時間來做別的事情了，一般在工作中我們都是程序+執行緒+協程的方式來實現併發，以達到最好的併發效果，如果是4核的cpu，一般起5個程序，每個程序中20個執行緒（5倍cpu數量），每個執行緒可以起500個協程，大規模爬取頁面的時候，等待網路延遲的時間的時候，我們就可以用協程去實現併發。 併發數量 = 5 * 20 * 500 = 50000個併發，這是一般一個4cpu的機器最大的併發數。nginx在負載均衡的時候最大承載量就是5w個
單執行緒裡的這20個任務的程式碼通常會既有計算操作又有阻塞操作，我們完全可以在執行任務1時遇到阻塞，就利用阻塞的時間去執行任務2。。。。如此，才能提高效率，這就用到了Gevent模組。
四.Gevent介紹
#安裝 pip3 install gevent
Gevent 是一個第三方庫，可以輕鬆通過gevent實現併發同步或非同步程式設計，在gevent中用到的主要模式是Greenlet, 它是以C擴充套件模組形式接入Python的輕量級協程。 Greenlet全部執行在主程式作業系統程序的內部，但它們被協作式地排程。
#用法 g1=gevent.spawn(func,1,2,3,x=4,y=5)建立一個協程物件g1，spawn括號內第一個引數是函式名，如eat，後面可以有多個引數，可以是位置實參或關鍵字實參，都是傳給函式eat的，spawn是非同步提交任務 g2=gevent.spawn(func2) g1.join() #等待g1結束，上面只是建立協程物件，這個join才是去執行 g2.join() #等待g2結束 有人測試的時候會發現，不寫第二個join也能執行g2，是的，協程幫你切換執行了，但是你會發現，如果g2裡面的任務執行的時間長，但是不寫join的話，就不會執行完等到g2剩下的任務了 #或者上述兩步合作一步：gevent.joinall([g1,g2]) g1.value#拿到func1的返回值
遇到IO阻塞時會自動切換任務
import gevent
def eat(name):
print('%s eat 1' %name)
gevent.sleep(2)
print('%s eat 2' %name)
def play(name):
print('%s play 1' %name)
gevent.sleep(1)
print('%s play 2' %name)
g1=gevent.spawn(eat,'egon')
g2=gevent.spawn(play,name='egon')
g1.join()
g2.join()
#或者gevent.joinall([g1,g2])
print('主')
上例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()放到檔案的開頭
from gevent import monkey;monkey.patch_all() #必須寫在最上面，這句話後面的所有阻塞全部能夠識別了
import gevent #直接匯入即可
import time
def eat():
#print()
print('eat food 1')
time.sleep(2) #加上mokey就能夠識別到time模組的sleep了
print('eat food 2')
def play():
print('play 1')
time.sleep(1) #來回切換，直到一個I/O的時間結束，這裡都是我們個gevent做得，不再是控制不了的作業系統了。
print('play 2')
g1=gevent.spawn(eat)
g2=gevent.spawn(play_phone)
gevent.joinall([g1,g2])
print('主')
我們可以用threading.current_thread().getName()來檢視每個g1和g2，檢視的結果為DummyThread-n，即假執行緒，虛擬執行緒，其實都在一個執行緒裡面
程序執行緒的任務切換是由作業系統自行切換的，你自己不能控制
協程是通過自己的程式（程式碼）來進行切換的，自己能夠控制，只有遇到協程模組能夠識別的IO操作的時候，程式才會進行任務切換，實現併發效果，如果所有程式都沒有IO操作，那麼就基本屬於序列執行了。
五.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.time()
gevent.joinall([
gevent.spawn(get_page,'https://www.python.org/'),
gevent.spawn(get_page,'https://www.yahoo.com/'),
gevent.spawn(get_page,'https://github.com/'),
])
stop_time=time.time()
print('run time is %s' %(stop_time-start_time))
將上面的程式最後加上一段序列的程式碼看看效率：如果你的程式不需要太高的效率，那就不用什麼併發啊協程啊之類的東西。
print('--------------------------------')
s = time.time()
requests.get('https://www.python.org/')
requests.get('https://www.yahoo.com/')
requests.get('https://github.com/')
t = time.time()
print('序列時間>>',t-s)
七.Gevent之應用舉例二
通過gevent實現單執行緒下的socket併發（from gevent import monkey;monkey.patch_all()一定要放到匯入socket模組之前，否則gevent無法識別socket的阻塞）
一個網路請求裡面經過多個時間延遲time
服務端
from gevent import monkey;monkey.patch_all()
from socket import *
import gevent
#如果不想用money.patch_all()打補丁,可以用gevent自帶的socket
# from gevent import socket
# s=socket.socket()
def server(server_ip,port):
s=socket(AF_INET,SOCK_STREAM)
s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
s.bind((server_ip,port))
s.listen(5)
while True:
conn,addr=s.accept()
gevent.spawn(talk,conn,addr)
def talk(conn,addr):
try:
while True:
res=conn.recv(1024)
print('client %s:%s msg: %s' %(addr[0],addr[1],res))
conn.send(res.upper())
except Exception as e:
print(e)
finally:
conn.close()
if __name__ == '__main__':
server('127.0.0.1',8080)
客戶端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
msg=input('>>: ').strip()
if not msg:continue
client.send(msg.encode('utf-8'))
msg=client.recv(1024)
多執行緒併發多個客戶端,去請求上面的服務端是沒問題的
from threading import Thread
from socket import *
import threading
def client(server_ip,port):
c=socket(AF_INET,SOCK_STREAM) #套接字物件一定要加到函式內，即區域性名稱空間內，放在函式外則被所有執行緒共享，則大家公用一個套接字物件，那麼客戶端埠永遠一樣了
c.connect((server_ip,port))
count=0
while True:
c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
msg=c.recv(1024)
print(msg.decode('utf-8'))
count+=1
if __name__ == '__main__':
for i in range(500):
t=Thread(target=client,args=('127.0.0.1',8080))
t.start()
今天的內容就到這裡