python協程例子

『壹』 python里怎麼實現多個協程一起執行，只要完

需要使用新的函數as_completed()來實現，可以把多個並發的協程一起給它，但它把返回的結果變成一個生成器，每次返回一個協程的結果，與函數wait()一樣，執行協程是亂序的，不會等所有協程執行完成才返回。例子：

importasyncio


asyncdefphase(i):
print('inphase{}'.format(i))
awaitasyncio.sleep(0.5-(0.1*i))
print('donewithphase{}'.format(i))
return'phase{}result'.format(i)


asyncdefmain(num_phases):
print('startingmain')
phases=[
phase(i)
foriinrange(num_phases)
]
print('waitingforphasestocomplete')
results=[]
fornext_to_completeinasyncio.as_completed(phases):
answer=awaitnext_to_complete
print('receivedanswer{!r}'.format(answer))
results.append(answer)
print('results:{!r}'.format(results))
returnresults


event_loop=asyncio.get_event_loop()
try:
event_loop.run_until_complete(main(3))
finally:
event_loop.close()

結果輸出如下：starting main
waiting for phases to complete
in phase 2
in phase 1
in phase 0
done with phase 2
received answer 'phase 2 result'
done with phase 1
received answer 'phase 1 result'
done with phase 0
received answer 'phase 0 result'
results: ['phase 2 result', 'phase 1 result', 'phase 0 result']

『貳』 Python協程之asyncio

asyncio 是 Python 中的非同步IO庫，用來編寫並發協程，適用於IO阻塞且需要大量並發的場景，例如爬蟲、文件讀寫。

asyncio 在 Python3.4 被引入，經過幾個版本的迭代，特性、語法糖均有了不同程度的改進，這也使得不同版本的 Python 在 asyncio 的用法上各不相同，顯得有些雜亂，以前使用的時候也是本著能用就行的原則，在寫法上走了一些彎路，現在對 Python3.7+ 和 Python3.6 中 asyncio 的用法做一個梳理，以便以後能更好的使用。

協程，又稱微線程，它不被操作系統內核所管理，而完全是由程序控制，協程切換花銷小，因而有更高的性能。

協程可以比作子程序，不同的是，執行過程中協程可以掛起當前狀態，轉而執行其他協程，在適當的時候返回來接著執行，協程間的切換不需要涉及任何系統調用或任何阻塞調用，完全由協程調度器進行調度。

Python 中以 asyncio 為依賴，使用 async/await 語法進行協程的創建和使用，如下 async 語法創建一個協程函數：

在協程中除了普通函數的功能外最主要的作用就是：使用 await 語法等待另一個協程結束，這將掛起當前協程，直到另一個協程產生結果再繼續執行：

asyncio.sleep() 是 asyncio 包內置的協程函數，這里模擬耗時的IO操作，上面這個協程執行到這一句會掛起當前協程而去執行其他協程，直到sleep結束，當有多個協程任務時，這種切換會讓它們的IO操作並行處理。

注意，執行一個協程函數並不會真正的運行它，而是會返回一個協程對象，要使協程真正的運行，需要將它們加入到事件循環中運行，官方建議 asyncio 程序應當有一個主入口協程，用來管理所有其他的協程任務：

在 Python3.7+ 中，運行這個 asyncio 程序只需要一句： asyncio.run(main()) ，而在 Python3.6 中，需要手動獲取事件循環並加入協程任務：

事件循環就是一個循環隊列，對其中的協程進行調度執行，當把一個協程加入循環，這個協程創建的其他協程都會自動加入到當前事件循環中。

其實協程對象也不是直接運行，而是被封裝成一個個待執行的 Task ，大多數情況下 asyncio 會幫我們進行封裝，我們也可以提前自行封裝 Task 來獲得對協程更多的控制權，注意，封裝 Task 需要 當前線程有正在運行的事件循環 ，否則將引 RuntimeError，這也就是官方建議使用主入口協程的原因，如果在主入口協程之外創建任務就需要先手動獲取事件循環然後使用底層方法 loop.create_task() ，而在主入口協程之內是一定有正在運行的循環的。任務創建後便有了狀態，可以查看運行情況，查看結果，取消任務等：

asyncio.create_task() 是 Python3.7 加入的高層級API，在 Python3.6，需要使用低層級API asyncio.ensure_future() 來創建 Future，Future 也是一個管理協程運行狀態的對象，與 Task 沒有本質上的區別。

通常，一個含有一系列並發協程的程序寫法如下（Python3.7+）：

並發運行多個協程任務的關鍵就是 asyncio.gather(*tasks) ，它接受多個協程任務並將它們加入到事件循環，所有任務都運行完成後會返回結果列表，這里我們也沒有手動封裝 Task，因為 gather 函數會自動封裝。

並發運行還有另一個方法 asyncio.wait(tasks) ，它們的區別是：

『叄』 詳解Python中的協程，為什麼說它的底層是生成器

協程又稱為是微線程，英文名是Coroutine。它和線程一樣可以調度，但是不同的是線程的啟動和調度需要通過操作系統來處理。並且線程的啟動和銷毀需要涉及一些操作系統的變數申請和銷毀處理，需要的時間比較長。而協程呢，它的調度和銷毀都是程序自己來控制的，因此它更加輕量級也更加靈活。

協程有這么多優點，自然也會有一些缺點，其中最大的缺點就是需要編程語言自己支持，否則的話需要開發者自己通過一些方法來實現協程。對於大部分語言來說，都不支持這一機制。go語言由於天然支持協程，並且支持得非常好，使得它廣受好評，短短幾年時間就迅速流行起來。

對於Python來說，本身就有著一個GIL這個巨大的先天問題。GIL是Python的全局鎖，在它的限制下一個Python進程同一時間只能同時執行一個線程，即使是在多核心的機器當中。這就大大影響了Python的性能，尤其是在CPU密集型的工作上。所以為了提升Python的性能，很多開發者想出了使用多進程+協程的方式。一開始是開發者自行實現的，後來在Python3.4的版本當中，官方也收入了這個功能，因此目前可以光明正大地說，Python是支持協程的語言了。

生成器（generator)

生成器我們也在之前的文章當中介紹過，為什麼我們介紹協程需要用到生成器呢，是因為Python的協程底層就是通過生成器來實現的。

通過生成器來實現協程的原因也很簡單，我們都知道協程需要切換掛起，而生成器當中有一個yield關鍵字，剛好可以實現這個功能。所以當初那些自己在Python當中開發協程功能的程序員都是通過生成器來實現的，我們想要理解Python當中協程的運用，就必須從最原始的生成器開始。

生成器我們很熟悉了，本質上就是帶有yield這個關鍵詞的函數。

async，await和future

從Python3.5版本開始，引入了async，await和future。我們來簡單說說它們各自的用途，其中async其實就是@asyncio.coroutine，用途是完全一樣的。同樣await代替的是yield from，意為等待另外一個協程結束。

我們用這兩個一改，上面的代碼就成了：

async def test(k):

n = 0

while n < k:

await asyncio.sleep(0.5)

print('n = {}'.format(n))

n += 1

由於我們加上了await，所以每次在列印之前都會等待0.5秒。我們把await換成yield from也是一樣的，只不過用await更加直觀也更加貼合協程的含義。

Future其實可以看成是一個信號量，我們創建一個全局的future，當一個協程執行完成之後，將結果存入這個future當中。其他的協程可以await future來實現阻塞。我們來看一個例子就明白了：

future = asyncio.Future()

async def test(k):

n = 0

while n < k:

await asyncio.sleep(0.5)

print('n = {}'.format(n))

n += 1

future.set_result('success')

async def log():

result = await future

print(result)

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait([

log(),

test(5)

]))

loop.close()

在這個例子當中我們創建了兩個協程，第一個協程是每隔0.5秒print一個數字，在print完成之後把success寫入到future當中。第二個協程就是等待future當中的數據，之後print出來。

在loop當中我們要調度執行的不再是一個協程對象了而是兩個，所以我們用asyncio當中的wait將這兩個對象包起來。只有當wait當中的兩個對象執行結束，wait才會結束。loop等待的是wait的結束，而wait等待的是傳入其中的協程的結束，這就形成了一個依賴循環，等價於這兩個協程對象結束，loop才會結束。

總結

async並不只是可以用在函數上，事實上還有很多其他的用法，比如用在with語句上，用在for循環上等等。這些用法比較小眾，細節也很多，就不一一展開了，大家感興趣的可以自行去了解一下。

不知道大家在讀這篇文章的過程當中有沒有覺得有些費勁，如果有的話，其實是很正常的。原因也很簡單，因為Python原生是不支持協程這個概念的，所以在一開始設計的時候也沒有做這方面的准備，是後來覺得有必要才加入的。那麼作為後面加入的內容，必然會對原先的很多內容產生影響，尤其是協程藉助了之前生成器的概念來實現的，那麼必然會有很多耦合不清楚的情況。這也是這一塊的語法很亂，對初學者不友好的原因。

『肆』 如何用python寫一個協程

我學習了asyncio的協程，現在在我的印象中一個協程有兩個要素：
* 用`asyncio.coroutine`裝飾
* 用`yield from`調用其他協程
我想要了解協程是什麼，所以做了以下嘗試。
我經過嘗試，發現運行構造出來的協程得到的是一個`generator`（迭代器）。
而最常規的迭代器生成使用的是`yield`。
所以同樣是生成迭代器，那協程是否可以用`yield`而不是`yield from`。
我經過嘗試，發現協程的調用有特殊的方式。
而最常規的迭代器都是直接調用就可以的。
所以，同樣是函數，那協程是否可以脫離`event_loop`（消息循環）調用。
我還嘗試過通過`yield`構造一個協程。
沒有報錯也運行成功了，所以應該沒有問題。

『伍』 python中的協程是怎麼實現多任務的

協程也稱為微線程，是在一個線程中，通過不斷的切換任務函數實現了多任務的效果。
協程在python實現的原理主要是通過yield這個關鍵字實現
但是真正在開發時，可以不需要自己實現，可以通過很多成熟的第三方模塊來實現協程，比如greenlet,gevent等模塊。黑馬程序員可學習Python哦，有免費的學習視頻，學習路線圖，學習工具！

『陸』 如何用python寫一個協程

是個很長的問題，我提供一下我理解的題主你的問題：
我學習了asyncio的協程，現在在我的印象中一個協程有兩個要素：
* 用`asyncio.coroutine`裝飾
* 用`yield from`調用其他協程
我想要了解協程是什麼，所以做了以下嘗試。
我經過嘗試，發現運行構造出來的協程得到的是一個`generator`（迭代器）。
而最常規的迭代器生成使用的是`yield`。
所以同樣是生成迭代器，那協程是否可以用`yield`而不是`yield from`。
我經過嘗試，發現協程的調用有特殊的方式。
而最常規的迭代器都是直接調用就可以的。
所以，同樣是函數，那協程是否可以脫離`event_loop`（消息循環）調用。
我還嘗試過通過`yield`構造一個協程。
沒有報錯也運行成功了，所以應該沒有問題。

『柒』 如何用python寫一個協程

作者：LittleCoder
鏈接：https://www.hu.com/question/54483694/answer/139785021
來源：知乎
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請註明出處。

yield`和`yield from`的區別

`yield`題主肯定不陌生，而`yield from`是PEP 380中新增的一個特性。
PEP 380的名字是嵌套子迭代器的語法糖（我喜歡這么翻譯，原文是：Syntax for Delegating to a Subgenerator）。
既然是語法糖，那麼肯定本來是有別的寫法的，這里給出本來的寫法：
def subgen():
for i in range(3):
yield 'subgen: %s' % i
return 'subgen returned'def gen():
r = yield from subgen()
print('r = "%s"' % r)
yield rdef gen_without_yield_from():
sg = subgen()
try:
while 1:
yield sg.send(None)
except StopIteration as e:
yield e.valueprint('* [gen] get all values')for v in gen_without_yield_from():
print('get value: %s' % v)print('* [gen_without_yield_from] get all values')for v in gen_without_yield_from():
print('get value: %s' % v)

不難看出，`yield`子迭代器是把子迭代器直接傳遞出去，`yield from`子迭代器是把子迭代器的值一個一個傳出去。
雖然實際把子迭代器當做一個對象直接傳遞出去也沒有問題，也有使用場景（生成迭代器的迭代器）。
但在協程中相較於這個令人愉快的語法糖而言，直接傳遞就顯得沒有必要且礙事了。
畢竟我希望使用一個子迭代器是把子迭代器中的代碼都運行一遍而不是直接就把這個子迭代器傳出來讓我自己操作。
所以如果你把子迭代器直接傳了出去，asyncio就判斷你在做一件奇怪的事情並報了錯。
那麼，回到問題，給出的程序要怎麼通過`yield`調用呢？
# 源程序@asyncio.coroutinedef hello():
print("Hello world!")
yield from asyncio.sleep(1)
print("Hello again!")# 使用[email protected] hello():
print("Hello world!")
for v in asyncio.sleep(1):
yield v
print("Hello again!")

協程和迭代器的區別

舉個比喻，迭代器和協程就像火葯和槍械，利用火葯的特性輔助各種其他東西才造出了槍械。
迭代器就最簡單的本質而言就是一個可以暫停的程序。
那麼就有這樣一個合理的聯想，我是不是可以節省下所有不必要的例如等待網站響應的等待時間。
就是我把我的請求發過去以後就把這個程序暫停下來，開啟別的程序，等到響應來了再叫我回到這個程序。
那麼等待網站響應的時間也就完全沒有浪費了，比原來傻傻的等著網站響應真是優秀了許多。
這就是協程。
所以，為什麼看上去都是`generator`，迭代器不會天生成為協程呢？
因為沒有一個知道什麼時候應該叫你回到這個程序的人。
這個人就是`event_loop`（消息循環）。
回到問題，協程是否可以脫離`event_loop`（消息循環）調用。
講道理是不可以的，但合理聯想一下是不是一直不停的告訴程序又到你了就行了。
像這樣：
@asyncio.coroutinedef gen():
for i in range(3):
yield ifor i in gen():
print(i)print('end')

的確有些協程這樣是可以運行的（這些協程為什麼要寫成協程？）。
但終究你是在不應該告訴程序到你的時候告訴了他這件事情。
所以顯然獲取數據的話當時數據根本沒有傳到，`sleep`的話就根本沒有了`sleep`的效果。
只是看上去能夠運行，實際完全沒有用。
asyncio還為此特地加了一個斷言，如果你這樣調用`asyncio.sleep`，asyncio會發現你在偽裝消息循環騙他。
協程的原理

這是另一個看上去能夠運行，實際上完全沒有用的事情。
這雖然不是你想問的問題，但你已經碰到了也遲早會意識到，所以一並講了。
這個問題應該是這樣的：為什麼我寫出來的協程完全沒有協程的效果？
import time, [email protected] sleep(symbol, i):
time.sleep(i)
print('[%s] finished')loop = asyncio.get_event_loop()tasks = [sleep('A', 2), sleep('B', 2)]loop.run_until_complete(asyncio.wait(tasks))loop.close()

看到這里你起碼可以簡單的講出來，因為顯然我們在傻傻的等。
我們沒有在開始等待的時候把程序暫停下來，然後在等待結束後繼續運行程序，我們一心一意的在等。
我們真的`time.sleep`了兩秒，而不是去做了兩秒其他的事情。
你有各種選擇，可以花式等待。我這里給你兩個最基本的例子：
* get請求
* 同步變為協程（線程池）
get請求
為了讓你更好的了解asyncio，我從最底層的socket開始寫一個協程的get請求給你。
為了模擬延時很大的網站，我在本地開了一個延時伺服器，這是伺服器程序。
import tornado.ioloopimport tornado.webfrom tornado.gen import coroutine, sleepclass MainHandler(tornado.web.RequestHandler):
@coroutine
def get(self, waitTime=3):
yield sleep(int(waitTime))
self.write('you have waited for %ss' % waitTime)if __name__ == "__main__":
application = tornado.web.Application([
('/([0-9])', MainHandler),
], debug=True)
application.listen(5000)
try:
tornado.ioloop.IOLoop.current().start()
except:
tornado.ioloop.IOLoop.current().stop()

記得打開了這個伺服器再運行下面的程序。
import socket, asyncio, timedata = 'GET /%s HTTP/1.1\r\n\r\n'loop = asyncio.get_event_loop()@asyncio.coroutinedef get(i):
future = asyncio.futures.Future(loop=loop)
s = socket.socket()
s.connect(('127.0.0.1', 5000))
s.sendall((data % i).encode('utf8'))
s.setblocking(False)
def callback(future):
future.set_result(s.recv(999).split(b'\r\n\r\n')[-1])
loop.add_reader(s.fileno(), callback, future)
r = yield from future
print('Return value: %s' % r)tasks = [get(3), get(3)]loop.run_until_complete(asyncio.wait(tasks))loop.close()

同步變為協程（線程池）
這里拿sleep模擬耗時的程序，原理就是開了5個新的線程處理耗時程序。
當然實際的`asyncio.sleep`只需要告訴消息循環一定時間後叫醒我就好了。
import asyncio, sleep, [email protected] sleep(i):
executor = concurrent.futures.ThreadPoolExecutor(5)
future = asyncio.futures.wrap_future(executor.submit(time.sleep, i), loop=loop)
yield from future
print('Slept for %s seconds' % i)tasks = [sleep(3), sleep(3)]loop.run_until_complete(asyncio.wait(tasks))loop.close()

『捌』 python中多進程+協程的使用以及為什麼要用它

python里推薦用多進程而不是多線程，但是多進程也有其自己的限制：相比線程更加笨重、切換耗時更長，並且在python的多進程下，進程數量不推薦超過CPU核心數（一個進程只有一個GIL，所以一個進程只能跑滿一個CPU），因為一個進程佔用一個CPU時能充分利用機器的性能，但是進程多了就會出現頻繁的進程切換，反而得不償失。
不過特殊情況（特指IO密集型任務）下，多線程是比多進程好用的。
舉個例子：給你200W條url，需要你把每個url對應的頁面抓取保存起來，這種時候，單單使用多進程，效果肯定是很差的。為什麼呢？
例如每次請求的等待時間是2秒，那麼如下（忽略cpu計算時間）：
1、單進程+單線程：需要2秒*200W=400W秒==1111.11個小時==46.3天，這個速度明顯是不能接受的
2、單進程+多線程：例如我們在這個進程中開了10個多線程，比1中能夠提升10倍速度，也就是大約4.63天能夠完成200W條抓取，請注意，這里的實際執行是：線程1遇見了阻塞，CPU切換到線程2去執行，遇見阻塞又切換到線程3等等，10個線程都阻塞後，這個進程就阻塞了，而直到某個線程阻塞完成後，這個進程才能繼續執行，所以速度上提升大約能到10倍（這里忽略了線程切換帶來的開銷，實際上的提升應該是不能達到10倍的），但是需要考慮的是線程的切換也是有開銷的，所以不能無限的啟動多線程（開200W個線程肯定是不靠譜的）
3、多進程+多線程：這里就厲害了，一般來說也有很多人用這個方法，多進程下，每個進程都能佔一個cpu，而多線程從一定程度上繞過了阻塞的等待，所以比單進程下的多線程又更好使了，例如我們開10個進程，每個進程里開20W個線程，執行的速度理論上是比單進程開200W個線程快10倍以上的（為什麼是10倍以上而不是10倍，主要是cpu切換200W個線程的消耗肯定比切換20W個進程大得多，考慮到這部分開銷，所以是10倍以上）。
還有更好的方法嗎？答案是肯定的，它就是：
4、協程，使用它之前我們先講講what/why/how（它是什麼/為什麼用它/怎麼使用它）
what：
協程是一種用戶級的輕量級線程。協程擁有自己的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其他地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。因此：
協程能保留上一次調用時的狀態（即所有局部狀態的一個特定組合），每次過程重入時，就相當於進入上一次調用的狀態，換種說法：進入上一次離開時所處邏輯流的位置。
在並發編程中，協程與線程類似，每個協程表示一個執行單元，有自己的本地數據，與其它協程共享全局數據和其它資源。
why：
目前主流語言基本上都選擇了多線程作為並發設施，與線程相關的概念是搶占式多任務（Preemptive multitasking），而與協程相關的是協作式多任務。
不管是進程還是線程，每次阻塞、切換都需要陷入系統調用(system call)，先讓CPU跑操作系統的調度程序，然後再由調度程序決定該跑哪一個進程(線程)。而且由於搶占式調度執行順序無法確定的特點，使用線程時需要非常小心地處理同步問題，而協程完全不存在這個問題（事件驅動和非同步程序也有同樣的優點）。
因為協程是用戶自己來編寫調度邏輯的，對CPU來說，協程其實是單線程，所以CPU不用去考慮怎麼調度、切換上下文，這就省去了CPU的切換開銷，所以協程在一定程度上又好於多線程。
how:
python裡面怎麼使用協程？答案是使用gevent，使用方法：看這里
使用協程，可以不受線程開銷的限制，我嘗試過一次把20W條url放在單進程的協程里執行，完全沒問題。
所以最推薦的方法，是多進程+協程（可以看作是每個進程里都是單線程，而這個單線程是協程化的）
多進程+協程下，避開了CPU切換的開銷，又能把多個CPU充分利用起來，這種方式對於數據量較大的爬蟲還有文件讀寫之類的效率提升是巨大的。

『玖』 python里協程事件循環里怎麼樣調用非協程函數

為了管理協程和I/O的回調函數，asyncio庫的事件循環也能基於定時的方式調用普通的函數，使用call_soon()函數，例子如下：

importasyncio
importfunctools
defcallback(arg,*,kwarg='default'):
print('callbackinvokedwith{}and{}'.format(arg,kwarg))
asyncdefmain(loop):
print('registeringcallbacks')
loop.call_soon(callback,1)
wrapped=functools.partial(callback,kwarg='notdefault')
loop.call_soon(wrapped,2)
awaitasyncio.sleep(0.1)
event_loop=asyncio.get_event_loop()
try:
print('enteringeventloop')
event_loop.run_until_complete(main(event_loop))
finally:
print('closingeventloop')
event_loop.close()

結果輸出如下：
entering event loop
registering callbacks
callback invoked with 1 and default
callback invoked with 2 and not default
closing event loop

『拾』 python里並發執行協程時部分阻塞超時怎麼辦

碰到這種需求時不要驚慌，可以使用wait()里的timeout參數來設置等待時間，也就是從這個函數開始運行算起，如果時間到達協程沒有執行完成，就可以不再等它們了，直接從wait()函數里返回，返回之後就可以判斷那些沒有執行成功的，可以把這些協程取消掉。例子如下

importasyncio


asyncdefphase(i):
print('inphase{}'.format(i))
try:
awaitasyncio.sleep(0.1*i)
exceptasyncio.CancelledError:
print('phase{}canceled'.format(i))
raise
else:
print('donewithphase{}'.format(i))
return'phase{}result'.format(i)


asyncdefmain(num_phases):
print('startingmain')
phases=[
phase(i)
foriinrange(num_phases)
]
print('waiting0.1forphasestocomplete')
completed,pending=awaitasyncio.wait(phases,timeout=0.1)
print('{}completedand{}pending'.format(
len(completed),len(pending),
))
#
#asweexitwithoutfinishingthem.
ifpending:
print('cancelingtasks')
fortinpending:
t.cancel()
print('exitingmain')


event_loop=asyncio.get_event_loop()
try:
event_loop.run_until_complete(main(3))
finally:
event_loop.close()

結果輸出如下：

starting main
waiting 0.1 for phases to complete
in phase 0
in phase 2
in phase 1
done with phase 0
1 completed and 2 pending
canceling tasks
exiting main
phase 1 canceled
phase 2 canceled