Python 并發(fā)編程之協(xié)程/異步IO

萬皇之皇 2018-02-08

展開全文

引言

隨著node.js的盛行，相信大家今年多多少少都聽到了異步編程這個概念,。Python社區(qū)雖然對于異步編程的支持相比其他語言稍顯遲緩,，但是也在Python3.4中加入了asyncio，在Python3.5上又提供了async/await語法層面的支持,，剛正式發(fā)布的Python3.6中asynico也已經(jīng)由臨時版改為了穩(wěn)定版,。下面我們就基于Python3.4+來了解一下異步編程的概念以及asyncio的用法。

什么是協(xié)程

通常在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)勢就是省去了多線程之間的切換開銷，獲得了更大的運行效率,。Python中的asyncio也是基于協(xié)程來進行實現(xiàn)的,。在進入asyncio之前我們先來了解一下Python中怎么通過生成器進行協(xié)程來實現(xiàn)并發(fā)。

example1

我們先來看一個簡單的例子來了解一下什么是協(xié)程(coroutine),，對生成器不了解的朋友建議先看一下Stackoverflow上面的這篇高票回答（http:///questions/231767/what-does-the-yield-keyword-do）,。

>>> def coroutine():
... reply = yield 'hello'
... yield reply
...
>>> c = coroutine()
>>> next(c)
'hello'
>>> c.send('world')
'world'

example2

下面這個程序我們要實現(xiàn)的功能就是模擬多個學(xué)生同時向一個老師提交作業(yè)，按照傳統(tǒng)的話我們或許要采用多線程/多進程,，但是這里我們可以采用生成器來實現(xiàn)協(xié)程用來模擬并發(fā),。

如果下面這個程序讀起來有點困難，可以直接跳到后面部分,，并不影響閱讀,，等你理解協(xié)程的本質(zhì)，回過頭來看就很簡單了,。

from collections import deque
def student(name, homeworks):
    for homework in homeworks.items():
        yield (name, homework[0], homework[1])  # 學(xué)生'生成'作業(yè)給老師
class Teacher(object):
    def __init__(self, students):
        self.students = deque(students)
    def handle(self):
        '''老師處理學(xué)生作業(yè)'''
        while len(self.students):
            student = self.students.pop()
            try:
                homework = next(student)
                print('handling', homework[0], homework[1], homework[2])
            except StopIteration:
                pass
            else:
                self.students.appendleft(student)

下面我們來調(diào)用一下這個程序,。

Teacher([
    student('Student1', {'math': '1+1=2', 'cs': 'operating system'}),
    student('Student2', {'math': '2+2=4', 'cs': 'computer graphics'}),
    student('Student3', {'math': '3+3=5', 'cs': 'compiler construction'})
]).handle()

這是輸出結(jié)果，我們僅僅只用了一個簡單的生成器就實現(xiàn)了并發(fā)(concurrence),，注意不是并行(parallel),，因為我們的程序僅僅是運行在一個單線程當中。

handling Student3 cs compiler construction
handling Student2 cs computer graphics
handling Student1 cs operating system
handling Student3 math 3+3=5
handling Student2 math 2+2=4
handling Student1 math 1+1=2

使用asyncio模塊實現(xiàn)協(xié)程

從Python3.4開始asyncio模塊加入到了標準庫,，通過asyncio我們可以輕松實現(xiàn)協(xié)程來完成異步IO操作,。

解釋一下下面這段代碼，我們創(chuàng)造了一個協(xié)程display_date(num, loop),，然后它使用關(guān)鍵字yield from來等待協(xié)程asyncio.sleep(2)的返回結(jié)果,。而在這等待的2s之間它會讓出CPU的執(zhí)行權(quán)，直到asyncio.sleep(2)返回結(jié)果,。

# coroutine.py
import asyncio
import datetime
@asyncio.coroutine  # 聲明一個協(xié)程
def display_date(num, loop):
    end_time = loop.time() + 10.0
    while True:
        print('Loop: {} Time: {}'.format(num, datetime.datetime.now()))
        if (loop.time() + 1.0) >= end_time:
            break
        yield from asyncio.sleep(2)  # 阻塞直到協(xié)程sleep(2)返回結(jié)果
loop = asyncio.get_event_loop()  # 獲取一個event_loop
tasks = [display_date(1, loop), display_date(2, loop)]
loop.run_until_complete(asyncio.gather(*tasks))  # '阻塞'直到所有的tasks完成
loop.close()

下面是運行結(jié)果,，注意到并發(fā)的效果沒有，程序從開始到結(jié)束只用大約10s,，而在這里我們并沒有使用任何的多線程/多進程代碼,。在實際項目中你可以將asyncio.sleep(secends)替換成相應(yīng)的IO任務(wù)，比如數(shù)據(jù)庫/磁盤文件讀寫等操作,。

ziwenxie :: ~ ? python coroutine.py
Loop: 1 Time: 2016-12-19 16:06:46.515329
Loop: 2 Time: 2016-12-19 16:06:46.515446
Loop: 1 Time: 2016-12-19 16:06:48.517613
Loop: 2 Time: 2016-12-19 16:06:48.517724
Loop: 1 Time: 2016-12-19 16:06:50.520005
Loop: 2 Time: 2016-12-19 16:06:50.520169
Loop: 1 Time: 2016-12-19 16:06:52.522452
Loop: 2 Time: 2016-12-19 16:06:52.522567
Loop: 1 Time: 2016-12-19 16:06:54.524889
Loop: 2 Time: 2016-12-19 16:06:54.525031
Loop: 1 Time: 2016-12-19 16:06:56.527713
Loop: 2 Time: 2016-12-19 16:06:56.528102

在Python3.5中為我們提供更直接的對協(xié)程的支持,，引入了async/await關(guān)鍵字，上面的代碼我們可以這樣改寫,，使用async代替了@asyncio.coroutine,，使用了await代替了yield from，這樣我們的代碼變得更加簡潔可讀。

import asyncio
import datetime
async def display_date(num, loop):  # 聲明一個協(xié)程
    end_time = loop.time() + 10.0
    while True:
        print('Loop: {} Time: {}'.format(num, datetime.datetime.now()))
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(2)  # 等同于yield from
loop = asyncio.get_event_loop()  # 獲取一個event_loop
tasks = [display_date(1, loop), display_date(2, loop)]
loop.run_until_complete(asyncio.gather(*tasks))  # '阻塞'直到所有的tasks完成
loop.close()

asyncio模塊詳解

開啟事件循環(huán)有兩種方法,，一種方法就是通過調(diào)用run_until_complete,，另外一種就是調(diào)用run_forever。run_until_complete內(nèi)置add_done_callback,，使用run_forever的好處是可以通過自己自定義add_done_callback,，具體差異請看下面兩個例子。

run_until_complete()

import asyncio
async def slow_operation(future):
    await asyncio.sleep(1)
    future.set_result('Future is done!')
loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(slow_operation(future))
print(loop.is_running())  # False
loop.run_until_complete(future)
print(future.result())
loop.close()

run_forever()

run_forever相比run_until_complete的優(yōu)勢是添加了一個add_done_callback,，可以讓我們在task(future)完成的時候調(diào)用相應(yīng)的方法進行后續(xù)處理,。

import asyncio
async def slow_operation(future):
    await asyncio.sleep(1)
    future.set_result('Future is done!')
def got_result(future):
    print(future.result())
    loop.stop()
loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(slow_operation(future))
future.add_done_callback(got_result)
try:
    loop.run_forever()
finally:
    loop.close()

這里還要注意一點，即使你調(diào)用了協(xié)程方法,，但是如果事件循環(huán)沒有開啟,，協(xié)程也不會執(zhí)行，參考官方文檔的描述,，我剛被坑過,。

Calling a coroutine does not start its code running – the coroutine object returned by the call doesn’t do anything until you schedule its execution. There are two basic ways to start it running: call await coroutine or yield from coroutine from another coroutine (assuming the other coroutine is already running!), or schedule its execution using the ensure_future() function or the AbstractEventLoop.create_task() method. Coroutines (and tasks) can only run when the event loop is running.

Call

call_soon()

import asyncio
def hello_world(loop):
print('Hello World')
loop.stop()
loop = asyncio.get_event_loop()
# Schedule a call to hello_world()
loop.call_soon(hello_world, loop)
# Blocking call interrupted by loop.stop()
loop.run_forever()
loop.close()

下面是運行結(jié)果，我們可以通過call_soon提前注冊我們的task,，并且也可以根據(jù)返回的Handle進行cancel,。

Hello World

call_later()

import asyncio
import datetime
def display_date(end_time, loop):
    print(datetime.datetime.now())
    if (loop.time() + 1.0) <>end_time:
        loop.call_later(1, display_date, end_time, loop)
    else:
        loop.stop()
loop = asyncio.get_event_loop()
# Schedule the first call to display_date()
end_time = loop.time() + 5.0
loop.call_soon(display_date, end_time, loop)
# Blocking call interrupted by loop.stop()
loop.run_forever()
loop.close()

改動一下上面的例子我們來看一下call_later的用法，注意這里并沒有像上面那樣使用while循環(huán)進行操作,，我們可以通過call_later來設(shè)置每隔1秒去調(diào)用display_date()方法,。

2016-12-24 19:17:13.421649
2016-12-24 19:17:14.422933
2016-12-24 19:17:15.424315
2016-12-24 19:17:16.425571
2016-12-24 19:17:17.426874

Chain coroutines

import asyncio
async def compute(x, y):
    print('Compute %s + %s ...' % (x, y))
    await asyncio.sleep(1.0)  # 協(xié)程compute不會繼續(xù)往下面執(zhí)行，直到協(xié)程sleep返回結(jié)果
    return x + y
async def print_sum(x, y):
    result = await compute(x, y)  # 協(xié)程print_sum不會繼續(xù)往下執(zhí)行,，直到協(xié)程compute返回結(jié)果
    print('%s + %s = %s' % (x, y, result))
loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

下面是輸出結(jié)果

ziwenxie :: ~ ? python chain.py
Compute 1 + 2 ...
1 + 2 = 3

在爬蟲中使用asyncio來實現(xiàn)異步IO

下面我們來通過一個簡單的例子來看一下怎么在Python爬蟲項目中使用asyncio,。by the way: 根據(jù)我有限的實驗結(jié)果，如果要充分發(fā)揮asynio的威力,，應(yīng)該使用aiohttp而不是requests,。而且也要合理使用concurrent.futures模塊提供的線程池/進程池，這一點我會在下一篇博文描述,。

import asyncio
import requests
async def spider(loop):
    # run_in_exectuor會返回一個Future,，而不是coroutine object
    future1 = loop.run_in_executor(None, requests.get, 'https://www./')
    future2 = loop.run_in_executor(None, requests.get, 'http:///')
    # 通過命令行可以發(fā)現(xiàn)上面兩個網(wǎng)絡(luò)IO在并發(fā)進行
    response1 = await future1  # 阻塞直到future1完成
    response2 = await future2  # 阻塞直到future2完成
    print(len(response1.text))
    print(len(response2.text))
    return 'done'
loop = asyncio.get_event_loop()
# If the argument is a coroutine object, it is wrapped by ensure_future().
result = loop.run_until_complete(spider(loop))
print(result)
loop.close()