python協程實例

⑴ 如何用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實現的原理主要是通過yield這個關鍵字實現
但是真正在開發時，可以不需要自己實現，可以通過很多成熟的第三方模塊來實現協程，比如greenlet,gevent等模塊。多線程的課程我記得是在黑馬程序員裡面找的，一套，還有資料。

⑶ 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]view plain

• #-*-coding=utf-8-*-

• importrequests

• importgevent

• fromgeventimportmonkey;monkey.patch_all()

• importsys

• reload(sys)

• sys.setdefaultencoding('utf8')

• deffetch(url):

• try:

• s=requests.Session()

• r=s.get(url,timeout=1)#在這里抓取頁面

• exceptException,e:

• printe

• return''

• defprocess_start(url_list):

• tasks=[]

• forurlinurl_list:

• tasks.append(gevent.spawn(fetch,url))

• gevent.joinall(tasks)#使用協程來執行

• deftask_start(filepath,flag=100000):#每10W條url啟動一個進程

• withopen(filepath,'r')asreader:#從給定的文件中讀取url

• url=reader.readline().strip()

• url_list=[]#這個list用於存放協程任務

• i=0#計數器，記錄添加了多少個url到協程隊列

• whileurl!='':

• i+=1

• url_list.append(url)#每次讀取出url，將url添加到隊列

• ifi==flag:#一定數量的url就啟動一個進程並執行

• p=Process(target=process_start,args=(url_list,))

• p.start()

• url_list=[]#重置url隊列

• i=0#重置計數器

• url=reader.readline().strip()

• ifurl_listnot[]:#若退出循環後任務隊列里還有url剩餘

• p=Process(target=process_start,args=(url_list,))#把剩餘的url全都放到最後這個進程來執行

• p.start()

• if__name__=='__main__':

• task_start('./testData.txt')#讀取指定文件





細心的同學會發現：上面的例子中隱藏了一個問題：進程的數量會隨著url數量的增加而不斷增加，我們在這里不使用進程池multiprocessing.Pool來控制進程數量的原因是multiprocessing.Pool和gevent有沖突不能同時使用，但是有興趣的同學可以研究一下gevent.pool這個協程池。

⑷ python協程gevent怎麼用

在學習gevent之前，你肯定要知道你學的這個東西是什麼。

官方描述gevent

gevent is a coroutine-based Python networking library that uses greenlet to provide a high-level synchronous API on top of the libev event loop.

翻譯：gevent是一個基於協程的Python網路庫。我們先理解這句，也是這次學習的重點——協程。

wiki描述協程

與子常式一樣，協程也是一種程序組件。相對子常式而言，協程更為一般和靈活，但在實踐中使用沒有子常式那樣廣泛。子常式的起始處是惟一的入口點，一旦退出即完成了子常式的執行，子常式的一個實例只會返回一次；協程可以通過yield來調用其它協程。通過yield方式轉移執行權的協程之間不是調用者與被調用者的關系，而是彼此對稱、平等的。協程允許多個入口點，可以在指定位置掛起和恢復執行。

沒看懂？沒關系，我也沒看懂，不過算是有點線索：子常式。

子常式

過程有兩種，一種叫子常式（Subroutine），通常叫Sub；另一種叫函數（Function）。底層實現機制是一樣的，區別在於，Sub只執行操作，沒有返回值；Function不但執行操作，並且有返回值。用過VB的應該會比較清楚這點。（原諒我用了網路）說到底子常式就是過程，我們一般叫它函數。

說到函數，我就想吐槽了，不明白為什麼要叫函數。很多時候我們寫一個函數是為了封裝、模塊化某個功能，它是一個功能、或者說是一個過程。因為它包含的是類似於流程圖那樣的具體邏輯，先怎樣做，然後怎樣做；如果遇到A情況則怎樣，如果遇到B情況又怎樣。個人覺得還是叫過程比較好，叫做函數就讓人很糾結了，難道因為回歸到底層還是計算問題，出於數學的角度把它稱為函數？這個略坑啊！為了符合大家的口味，我還是稱之為函數好了（其實我也習慣叫函數了%>_

講到函數，我們就往底層深入一點，看看下面的代碼：

Python

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

def a():

print "a start"

b()

print "a end"

def b():

print "b start"

c()

print "b end"

def c():

print "c start"

print "c end"

if __name__ == "__main__":

a()

a start

b start

c start

c end

b end

a end

對於這樣的結果大家肯定不會意外的。每當函數被調用，就會在棧中開辟一個棧空間，調用結束後再回收該空間。

假設一個這樣的場景：有個講台，每個人都可以上去發表言論，但是每次講台只能站一個人。現在a在上面演講，當他說到「大家好！」的時候，b有個緊急通知要告訴大家，所以a就先下來讓b講完通知，然後a再上講台繼續演講。如果用函數的思想模擬這個問題，堆棧示意圖是這樣的：

那什麼東西有這樣的能力呢？我們很快就可以想到進程、線程，但是你真的想使用進程、線程如此重量級的東西在這么簡單的程序上嗎？野蠻的搶占式機制和笨重的上下文切換！

還有一種程序組件，那就是協程。它能保留上一次調用時的狀態，每次重新進入該過程的時候，就相當於回到上一次離開時所處邏輯流的位置。協程的起始處是第一個入口點，在協程里，返回點之後是接下來的入口點。協程的生命期完全由他們的使用的需要決定。每個協程在用yield命令向另一個協程交出控制時都盡可能做了更多的工作，放棄控制使得另一個協程從這個協程停止的地方開始，接下來的每次協程被調用時，都是從協程返回（或yield）的位置接著執行。

從上面這些你就可以知道其實協程是模擬了多線程（或多進程）的操作，多線程在切換的時候都會有一個上下文切換，在退出的時候將現場保存起來，等到下一次進入的時候從保存的現場開始，繼續執行。

看下協程是怎樣實現的：

Python

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

import random

from time import sleep

from greenlet import greenlet

from Queue import Queue

queue = Queue(1)

@greenlet

def procer():

chars = ['a', 'b', 'c', 'd', 'e']

global queue

while True:

char = random.choice(chars)

queue.put(char)

print "Proced: ", char

sleep(1)

consumer.switch()

@greenlet

def consumer():

global queue

while True:

char = queue.get()

print "Consumed: ", char

sleep(1)

procer.switch()

if __name__ == "__main__":

procer.run()

consumer.run()

應用場景

我們一直都在大談協程是什麼樣一個東西，卻從沒有提起協程用來幹嘛，這個其實大家分析一下就能夠知道。從上面的生產者——消費者問題應該能看出，它分別有兩個任務，假設交給兩個人去執行，但每次只能允許一個人行動。當緩沖區滿的時候，生產者是出於等待狀態的，這個時候可以將執行任務的權利轉交給消費者，當緩沖區空得時候，消費者是出於等待狀態的，這個時候可以將執行任務的權利轉交給生產者，是不是很容易聯想到多任務切換？然後想到線程？最後想到高並發？

但同學們又會問，既然有了線程為什麼還要協程呢？因為線程是系統級別的，在做切換的時候消耗是特別大的，具體為什麼這么大等我研究好了再告訴你；同時線程的切換是由CPU決定的，可能你剛好執行到一個地方的時候就要被迫終止，這個時候你需要用各種措施來保證你的數據不出錯，所以線程對於數據安全的操作是比較復雜的。而協程是用戶級別的切換，且切換是由自己控制，不受外力終止。

總結

協程其實模擬了人類活動的一種過程。例如：你准備先寫文檔，然後修復bug。這時候接到電話說這個bug很嚴重，必須立即修復（可以看作CPU通知）。於是你暫停寫文檔，開始去填坑，終於你把坑填完了，你回來寫文檔，這個時候你肯定是接著之前寫的文檔繼續，難道你要把之前寫的給刪了，重新寫？這就是協程。那如果是子常式呢？那你就必須重新寫了，因為退出之後，棧幀就會被彈出銷毀，再次調用就是開辟新的棧空間了。

總結：協程就是用戶態下的線程，是人們在有了進程、線程之後仍覺得效率不夠，而追求的又一種高並發解決方案。為什麼說是用戶態，是因為操作系統並不知道它的存在，它是由程序員自己控制、互相協作的讓出控制權而不是像進程、線程那樣由操作系統調度決定是否讓出控制權。

⑸ python中的協程是怎麼實現多任務的

協程也稱為微線程，是在一個線程中，通過不斷的切換任務函數實現了多任務的效果。
協程在python實現的原理主要是通過yield這個關鍵字實現
但是真正在開發時，可以不需要自己實現，可以通過很多成熟的第三方模塊來實現協程，比如greenlet,gevent等模塊。黑馬程序員可學習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里怎麼實現多個協程一起執行，只要完成

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