python2協程_python中的協程（二）

協程

1、協程：

單線程實現并發

在應用程序里控制多個任務的切換+保存狀態

優點：

應用程序級別速度要遠遠高于操作系統的切換

缺點：

多個任務一旦有一個阻塞沒有切，整個線程都阻塞在原地，該線程內的其他的任務都不能執行了

一旦引入協程，就需要檢測單線程下所有的IO行為, 實現遇到IO就切換,少一個都不行，以為一旦一個任務阻塞了，整個線程就阻塞了，其他的任務即便是可以計算，但是也無法運行了

2、協程序的目的：

想要在單線程下實現并發

并發指的是多個任務看起來是同時運行的

并發=切換+保存狀態

協程引入

cpu正在運行一個任務，會在兩種情況下切走去執行其他的任務(切換由操作系統強制控制)，一種情況是該任務發生了阻塞，另外一種情況是該任務計算的時間過長或有一個優先級更高的程序替代了它

ps：在介紹進程理論時，提及進程的三種執行狀態，而線程才是執行單位，所以也可以將上圖理解為線程的三種狀態

一：其中第二種情況并不能提升效率，只是為了讓cpu能夠雨露均沾，實現看起來所有任務都被“同時”執行的效果，如果多個任務都是純計算的，這種切換反而會降低效率。為此我們可以基于yield來驗證。yield本身就是一種在單線程下可以保存任務運行狀態的方法，我們來簡單復習一下：

#串行執行

import time

def func1():

for i in range(10000000):

i+1

def func2():

for i in range(10000000):

i+1

start = time.time()

func1()

func2()

stop = time.time()

print(stop - start)

#基于yield并發執行

import time

def func1():

while True:

yield

def func2():

g=func1()

for i in range(10000000):

i+1

next(g)

start=time.time()

func2()

stop=time.time()

print(stop-start)

二：第一種情況的切換。在任務一遇到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)

yield不能檢測IO，實現遇到IO自動切換

對于單線程下，我們不可避免程序中出現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

from greenlet import greenlet

def eat(name):

print('%s eat 1' %name)

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時傳入參數，以后都不需要

單純的切換(在沒有io的情況下或者沒有重復開辟內存空間的操作)，反而會降低程序的執行速度

greenlet只是提供了一種比generator更加便捷的切換方式，當切到一個任務執行時如果遇到io，那就原地阻塞，仍然是沒有解決遇到IO自動切換來提升效率的問題。

單線程里的這20個任務的代碼通常會既有計算操作又有阻塞操作，我們完全可以在執行任務1時遇到阻塞，就利用阻塞的時間去執行任務2。。。。如此，才能提高效率，這就用到了Gevent模塊(后續)。

greenlet module與class

我們首先看一下greenlet這個module里面的屬性

>> dir(greenlet)

['GREENLET_USE_GC', 'GREENLET_USE_TRACING', 'GreenletExit', '_C_API', '__doc__', '__file__', '__name__', '__package__', '__version__', 'error', 'getcurrent', 'gettrace', 'greenlet', 'settrace']

>>>

其中，比較重要的是getcurrent()，類greenlet、異常類GreenletExit。

getcurrent()返回當前的greenlet實例；

GreenletExit：是一個特殊的異常，當觸發了這個異常的時候，即使不處理，也不會拋到其parent(后面會提到協程中對返回值或者異常的處理)

然后我們再來看看greenlet.greenlet這個類：

>>> dir(greenlet.greenlet)

['GreenletExit', '__class__', '__delattr__', '__dict__', '__doc__', '__format__', '__getattribute__', '__getstate__', '__hash__', '__init__', '__new__', '__nonzero__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', 　　'__sizeof__', '__str__', '__subclasshook__', '_stack_saved', 'dead', 'error', 'getcurrent', 'gettrace', 'gr_frame', 'parent', 'run', 'settrace','switch', 'throw']

>>>

比較重要的幾個屬性：

run：當greenlet啟動的時候會調用到這個callable，如果我們需要繼承greenlet.greenlet時，需要重寫該方法

switch：前面已經介紹過了，在greenlet之間切換

parent：可讀寫屬性，后面介紹

dead：如果greenlet執行結束，那么該屬性為true

throw：切換到指定greenlet后立即跑出異常

文章后面提到的greenlet大多都是指greenlet.greenlet這個class，請注意區別。

parent屬性

每一個Greenlet都有一個parent，一個新的greenlet在哪里創生，當前環境的greenlet就是這個新greenlet的parent。所有的greenlet構成一棵樹，其跟節點就是還沒有手動創建greenlet時候的”main” greenlet(事實上，在首次import greenlet的時候實例化)。當一個協程正常結束，執行流程回到其對應的parent；或者在一個協程中拋出未被捕獲的異常，該異常也是傳遞到其parent。

import greenlet

def test1(x, y):

print id(greenlet.getcurrent()), id(greenlet.getcurrent().parent) # 40240272 40239952

z = gr2.switch(x+y)

print 'back z', z

def test2(u):

print id(greenlet.getcurrent()), id(greenlet.getcurrent().parent) # 40240352 40239952

return 'hehe'

gr1 = greenlet.greenlet(test1)

gr2 = greenlet.greenlet(test2)

print id(greenlet.getcurrent()), id(gr1), id(gr2) # 40239952, 40240272, 40240352

print gr1.switch("hello", " world"), 'back to main' # hehe back to main

上述例子可以看到，盡量是從test1所在的協程gr1 切換到了gr2，但gr2的parent還是’main’ greenlet，因為默認的parent取決于greenlet的創生環境。另外在test2中return之后整個返回值返回到了其parent，而不是switch到該協程的地方(即不是test1)，這個跟我們平時的函數調用不一樣，記住“switch not call”。對于異常也是展開至parent。

mport greenlet

def test1(x, y):

try:

z = gr2.switch(x+y)

except Exception:

print 'catch Exception in test1'

def test2(u):

assert False

gr1 = greenlet.greenlet(test1)

gr2 = greenlet.greenlet(test2)

try:

gr1.switch("hello", " world")

except:

print 'catch Exception in main'

輸出為：

catch Exception in main

Greenlet生命周期

文章開始的地方提到第一個例子中的gr2其實并沒有正常結束，我們可以借用greenlet.dead這個屬性來查看：

from greenlet import greenlet

def test1():

gr2.switch(1)

print 'test1 finished'

def test2(x):

print 'test2 first', x

z = gr1.switch()

print 'test2 back', z

gr1 = greenlet(test1)

gr2 = greenlet(test2)

gr1.switch()

print 'gr1 is dead?: %s, gr2 is dead?: %s' % (gr1.dead, gr2.dead)

gr2.switch()

print 'gr1 is dead?: %s, gr2 is dead?: %s' % (gr1.dead, gr2.dead)

print gr2.switch(10)

輸出：

test2 first 1

test1 finished

gr1 is dead?: True, gr2 is dead?: False

test2 back ()

gr1 is dead?: True, gr2 is dead?: True

從這個例子可以看出：

只有當協程對應的函數執行完畢，協程才會die，所以第一次Check的時候gr2并沒有die，因為第9行切換出去了就沒切回來。在main中再switch到gr2的時候，執行后面的邏輯，gr2 die。

如果試圖再次switch到一個已經是dead狀態的greenlet會怎么樣呢，事實上會切換到其parent greenlet。

Greenlet Traceing

Greenlet也提供了接口使得程序員可以監控greenlet的整個調度流程。主要是gettrace 和 settrace(callback)函數。下面看一個例子：

def test_greenlet_tracing():

def callback(event, args):

print event, 'from', id(args[0]), 'to', id(args[1])

def dummy():

g2.switch()

def dummyexception():

raise Exception('excep in coroutine')

main = greenlet.getcurrent()

g1 = greenlet.greenlet(dummy)

g2 = greenlet.greenlet(dummyexception)

print 'main id %s, gr1 id %s, gr2 id %s' % (id(main), id(g1), id(g2))

oldtrace = greenlet.settrace(callback)

try:

g1.switch()

except:

print 'Exception'

finally:

greenlet.settrace(oldtrace)

test_greenlet_tracing()

輸出：

main id 40604416, gr1 id 40604736, gr2 id 40604816

switch from 40604416 to 40604736

switch from 40604736 to 40604816

throw from 40604816 to 40604416

Exception

其中callback函數event是switch或者throw之一，表明是正常調度還是異常跑出；args是二元組，表示是從協程args[0]切換到了協程args[1]。上面的輸出展示了切換流程：從main到gr1，然后到gr2，最后回到main。

Greenlet內存泄漏問題

使用greenlet需要注意一下三點：

第一：greenlet創生之后，一定要結束，不能switch出去就不回來了，否則容易造成內存泄露。

第二：python中每個線程都有自己的main greenlet及其對應的sub-greenlet ，不能線程之間的greenlet是不能相互切換的。

第三：不能存在循環引用，這個是官方文檔明確說明。

”Greenlets do not participate in garbage collection; cycles involving data that is present in a greenlet’s frames will not be detected. “

from greenlet import greenlet, GreenletExit

huge = []

def show_leak():

def test1():

gr2.switch()

def test2():

huge.extend([x* x for x in range(100)])

gr1.switch()

print 'finish switch del huge'

del huge[:]

gr1 = greenlet(test1)

gr2 = greenlet(test2)

gr1.switch()

gr1 = gr2 = None

print 'length of huge is zero ? %s' % len(huge)

if __name__ == '__main__':

show_leak()

# output： length of huge is zero ? 100

在test2函數中第11行，我們將huge清空，然后再第16行將gr1、gr2的引用計數降到了0。但運行結果告訴我們，第11行并沒有執行，所以如果一個協程沒有正常結束是很危險的，往往不符合程序員的預期。greenlet提供了解決這個問題的辦法，官網文檔提到：如果一個greenlet實例的引用計數變成0，那么會在上次掛起的地方拋出GreenletExit異常，這就使得我們可以通過try … finally 處理資源泄露的情況。如下面的代碼：

from greenlet import greenlet, GreenletExit

huge = []

def show_leak():

def test1():

gr2.switch()

def test2():

huge.extend([x* x for x in range(100)])

try:

gr1.switch()

finally:

print 'finish switch del huge'

del huge[:]

gr1 = greenlet(test1)

gr2 = greenlet(test2)

gr1.switch()

gr1 = gr2 = None

print 'length of huge is zero ? %s' % len(huge)

if __name__ == '__main__':

show_leak()

# output :

# finish switch del huge

# length of huge is zero ? 0

上述代碼的switch流程：main greenlet –> gr1 –> gr2 –> gr1 –> main greenlet, 很明顯gr2沒有正常結束(在第10行刮起了)。第18行之后gr1,gr2的引用計數都變成0，那么會在第10行拋出GreenletExit異常，因此finally語句有機會執行。同時，在文章開始介紹Greenlet module的時候也提到了，GreenletExit這個異常并不會拋出到parent，所以main greenlet也不會出異常。

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的

g2=gevent.spawn(func2)

g1.join() #等待g1結束

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('eat food 1')

time.sleep(2)

print('eat food 2')

def play():

print('play 1')

time.sleep(1)

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，即假線程

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([