Python如何協(xié)程

Python如何協(xié)程？很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學(xué)習(xí)下，希望你能有所收獲。

通常在Python中我們進行并發(fā)編程一般都是使用多線程或者多進程來實現(xiàn)的，對于計算型任務(wù)由于GIL的存在我們通常使用多進程來實現(xiàn)，而對于IO型任務(wù)我們可以通過線程調(diào)度來讓線程在執(zhí)行IO任務(wù)時讓出GIL，從而實現(xiàn)表面上的并發(fā)。其實對于IO型任務(wù)我們還有一種選擇就是協(xié)程，協(xié)程是運行在單線程當中的"并發(fā)"，協(xié)程相比多線程一大優(yōu)勢就是省去了多線程之間的切換開銷，獲得了更大的運行效率。

協(xié)程，又稱微線程，纖程，英文名Coroutine。協(xié)程的作用是在執(zhí)行函數(shù)A時可以隨時中斷去執(zhí)行函數(shù)B，然后中斷函數(shù)B繼續(xù)執(zhí)行函數(shù)A（可以自由切換）。但這一過程并不是函數(shù)調(diào)用，這一整個過程看似像多線程，然而協(xié)程只有一個線程執(zhí)行。

那協(xié)程有什么優(yōu)勢呢？

執(zhí)行效率極高，因為子程序切換（函數(shù)）不是線程切換，由程序自身控制，沒有切換線程的開銷。所以與多線程相比，線程的數(shù)量越多，協(xié)程性能的優(yōu)勢越明顯。

不需要多線程的鎖機制，因為只有一個線程，也不存在同時寫變量沖突，在控制共享資源時也不需要加鎖，因此執(zhí)行效率高很多。

協(xié)程可以處理IO密集型程序的效率問題，但是處理CPU密集型不是它的長處，如要充分發(fā)揮CPU利用率可以結(jié)合多進程+協(xié)程。

Python中的協(xié)程經(jīng)歷了很長的一段發(fā)展歷程。其大概經(jīng)歷了如下三個階段：

最初的生成器變形yield/send

引入@asyncio.coroutine和yield from

引入async/await關(guān)鍵字

上述是協(xié)程概念和優(yōu)勢的一些簡介，感覺會比較抽象，Python2.x對協(xié)程的支持比較有限，生成器yield實現(xiàn)了一部分但不完全，gevent模塊倒是有比較好的實現(xiàn)；Python3.4加入了asyncio模塊，在Python3.5中又提供了async/await語法層面的支持，Python3.6中asyncio模塊更加完善和穩(wěn)定。接下來我們圍繞這些內(nèi)容詳細闡述一下。

Python2.x協(xié)程

python2.x實現(xiàn)協(xié)程的方式有：

yield + send

gevent (見后續(xù)章節(jié))

yield + send（利用生成器實現(xiàn)協(xié)程）

我們通過“生產(chǎn)者-消費者”模型來看一下協(xié)程的應(yīng)用，生產(chǎn)者生產(chǎn)消息后，直接通過yield跳轉(zhuǎn)到消費者開始執(zhí)行，待消費者執(zhí)行完畢后，切換回生產(chǎn)者繼續(xù)生產(chǎn)。

#-*- coding:utf8 -*-
def consumer():
    r = ''
    while True:
    n = yield r
    if not n:
        return
    print('[CONSUMER]Consuming %s...' % n)
    r = '200 OK'
def producer(c):
    # 啟動生成器
    c.send(None)
    n = 0
    while n < 5:
    n = n + 1
    print('[PRODUCER]Producing %s...' % n)
    r = c.send(n)
    print('[PRODUCER]Consumer return: %s' % r)
    c.close()
if __name__ == '__main__':
    c = consumer()
    producer(c)

復(fù)制代碼send(msg)與next()的區(qū)別在于send可以傳遞參數(shù)給yield表達式，這時傳遞的參數(shù)會作為yield表達式的值，而yield的參數(shù)是返回給調(diào)用者的值。換句話說，就是send可以強行修改上一個yield表達式的值。比如函數(shù)中有一個yield賦值a = yield 5，第一次迭代到這里會返回5，a還沒有賦值。第二次迭代時，使用send(10)，那么就是強行修改yield 5表達式的值為10，本來是5的，結(jié)果a = 10。send(msg)與next()都有返回值，它們的返回值是當前迭代遇到y(tǒng)ield時，yield后面表達式的值，其實就是當前迭代中yield后面的參數(shù)。第一次調(diào)用send時必須是send(None)，否則會報錯，之所以為None是因為這時候還沒有一個yield表達式可以用來賦值。上述例子運行之后輸出結(jié)果如下：

[PRODUCER]Producing 1...
[CONSUMER]Consuming 1...
[PRODUCER]Consumer return: 200 OK
[PRODUCER]Producing 2...
[CONSUMER]Consuming 2...
[PRODUCER]Consumer return: 200 OK
[PRODUCER]Producing 3...
[CONSUMER]Consuming 3...
[PRODUCER]Consumer return: 200 OK
[PRODUCER]Producing 4...
[CONSUMER]Consuming 4...
[PRODUCER]Consumer return: 200 OK
[PRODUCER]Producing 5...
[CONSUMER]Consuming 5...
[PRODUCER]Consumer return: 200 OK

Python3.x協(xié)程

除了Python2.x中協(xié)程的實現(xiàn)方式，Python3.x還提供了如下方式實現(xiàn)協(xié)程：

asyncio + yield from (python3.4+)

asyncio + async/await (python3.5+)

Python3.4以后引入了asyncio模塊，可以很好的支持協(xié)程。

asyncio + yield from

asyncio是Python3.4版本引入的標準庫，直接內(nèi)置了對異步IO的支持。asyncio的異步操作，需要在coroutine中通過yield from完成?？慈缦麓a（需要在Python3.4以后版本使用）：

#-*- coding:utf8 -*-
import asyncio
@asyncio.coroutine
def test(i):
    print('test_1', i)
    r = yield from asyncio.sleep(1)
    print('test_2', i)
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    tasks = [test(i) for i in range(3)]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

@asyncio.coroutine把一個generator標記為coroutine類型，然后就把這個coroutine扔到EventLoop中執(zhí)行。test()會首先打印出test_1，然后yield from語法可以讓我們方便地調(diào)用另一個generator。由于asyncio.sleep()也是一個coroutine，所以線程不會等待asyncio.sleep()，而是直接中斷并執(zhí)行下一個消息循環(huán)。當asyncio.sleep()返回時，線程就可以從yield from拿到返回值（此處是None），然后接著執(zhí)行下一行語句。把asyncio.sleep(1)看成是一個耗時1秒的IO操作，在此期間主線程并未等待，而是去執(zhí)行EventLoop中其他可以執(zhí)行的coroutine了，因此可以實現(xiàn)并發(fā)執(zhí)行。

asyncio + async/await

為了簡化并更好地標識異步IO，從Python3.5開始引入了新的語法async和await，可以讓coroutine的代碼更簡潔易讀。請注意，async和await是coroutine的新語法，使用新語法只需要做兩步簡單的替換：

把@asyncio.coroutine替換為async

把yield from替換為await

看如下代碼（在Python3.5以上版本使用）：

#-*- coding:utf8 -*-
import asyncio
async def test(i):
    print('test_1', i)
    await asyncio.sleep(1)
    print('test_2', i)
    
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    tasks = [test(i) for i in range(3)]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

運行結(jié)果與之前一致。與前一節(jié)相比，這里只是把yield from換成了await，@asyncio.coroutine換成了async，其余不變。

Gevent

Gevent是一個基于Greenlet實現(xiàn)的網(wǎng)絡(luò)庫，通過greenlet實現(xiàn)協(xié)程。基本思想是一個greenlet就認為是一個協(xié)程，當一個greenlet遇到IO操作的時候，比如訪問網(wǎng)絡(luò)，就會自動切換到其他的greenlet，等到IO操作完成，再在適當?shù)臅r候切換回來繼續(xù)執(zhí)行。由于IO操作非常耗時，經(jīng)常使程序處于等待狀態(tài)，有了gevent為我們自動切換協(xié)程，就保證總有g(shù)reenlet在運行，而不是等待IO操作。

Greenlet是作為一個C擴展模塊，它封裝了libevent事件循環(huán)的API，可以讓開發(fā)者在不改變編程習(xí)慣的同時，用同步的方式寫異步IO的代碼。

#-*- coding:utf8 -*-
import gevent
def test(n):
    for i in range(n):
        print(gevent.getcurrent(), i)
if __name__ == '__main__':
    g1 = gevent.spawn(test, 3)
    g2 = gevent.spawn(test, 3)
    g3 = gevent.spawn(test, 3)
    
    g1.join()
    g2.join()
    g3.join()

運行結(jié)果：

<Greenlet at 0x10a6eea60: test(3)> 0
<Greenlet at 0x10a6eea60: test(3)> 1
<Greenlet at 0x10a6eea60: test(3)> 2
<Greenlet at 0x10a6eed58: test(3)> 0
<Greenlet at 0x10a6eed58: test(3)> 1
<Greenlet at 0x10a6eed58: test(3)> 2
<Greenlet at 0x10a6eedf0: test(3)> 0
<Greenlet at 0x10a6eedf0: test(3)> 1
<Greenlet at 0x10a6eedf0: test(3)> 2

復(fù)制代碼可以看到3個greenlet是依次運行而不是交替運行。要讓greenlet交替運行，可以通過gevent.sleep()交出控制權(quán)：

def test(n):
    for i in range(n):
        print(gevent.getcurrent(), i)
        gevent.sleep(1)

運行結(jié)果：

<Greenlet at 0x10382da60: test(3)> 0
<Greenlet at 0x10382dd58: test(3)> 0
<Greenlet at 0x10382ddf0: test(3)> 0
<Greenlet at 0x10382da60: test(3)> 1
<Greenlet at 0x10382dd58: test(3)> 1
<Greenlet at 0x10382ddf0: test(3)> 1
<Greenlet at 0x10382da60: test(3)> 2
<Greenlet at 0x10382dd58: test(3)> 2
<Greenlet at 0x10382ddf0: test(3)> 2

當然在實際的代碼里，我們不會用gevent.sleep()去切換協(xié)程，而是在執(zhí)行到IO操作時gevent會自動完成，所以gevent需要將Python自帶的一些標準庫的運行方式由阻塞式調(diào)用變?yōu)閰f(xié)作式運行。這一過程在啟動時通過monkey patch完成：

#-*- coding:utf8 -*-
from gevent import monkey; monkey.patch_all()
from urllib import request
import gevent
def test(url):
    print('Get: %s' % url)
    response = request.urlopen(url)
    content = response.read().decode('utf8')
    print('%d bytes received from %s.' % (len(content), url))
    
if __name__ == '__main__':
    gevent.joinall([
    gevent.spawn(test, 'http://httpbin.org/ip'),
    gevent.spawn(test, 'http://httpbin.org/uuid'),
    gevent.spawn(test, 'http://httpbin.org/user-agent')
    ])

運行結(jié)果：

Get: http://httpbin.org/ip
Get: http://httpbin.org/uuid
Get: http://httpbin.org/user-agent
53 bytes received from http://httpbin.org/uuid.
40 bytes received from http://httpbin.org/user-agent.
31 bytes received from http://httpbin.org/ip.

從結(jié)果看，3個網(wǎng)絡(luò)操作是并發(fā)執(zhí)行的，而且結(jié)束順序不同，但只有一個線程。

總結(jié)

至此Python中的協(xié)程就介紹完畢了，示例程序中都是以sleep代表異步IO的，在實際項目中可以使用協(xié)程異步的讀寫網(wǎng)絡(luò)、讀寫文件、渲染界面等，而在等待協(xié)程完成的同時，CPU還可以進行其他的計算，協(xié)程的作用正在于此。那么協(xié)程和多線程的差異在哪里呢？多線程的切換需要靠操作系統(tǒng)來完成，當線程越來越多時切換的成本會很高，而協(xié)程是在一個線程內(nèi)切換的，切換過程由我們自己控制，因此開銷小很多，這就是協(xié)程和多線程的根本差異。

看完上述內(nèi)容是否對您有幫助呢？如果還想對相關(guān)知識有進一步的了解或閱讀更多相關(guān)文章，請關(guān)注億速云行業(yè)資訊頻道，感謝您對億速云的支持。