Python中協程用法代碼詳解

2020-01-04 15:53:02

字體：大中小

來源：轉載

供稿：網友

本文研究的主要是python/293574.html">python/57550.html">python中協程的相關問題，具體介紹如下。

Num01–>協程的定義

協程，又稱微線程，纖程。英文名Coroutine。

首先我們得知道協程是啥？協程其實可以認為是比線程更小的執行單元。為啥說他是一個執行單元，因為他自帶CPU上下文。這樣只要在合適的時機，我們可以把一個協程切換到另一個協程。只要這個過程中保存或恢復 CPU上下文那么程序還是可以運行的。

Num02–>協程和線程的差異

那么這個過程看起來和線程差不多。其實不然, 線程切換從系統層面遠不止保存和恢復 CPU上下文這么簡單。操作系統為了程序運行的高效性每個線程都有自己緩存Cache等等數據，操作系統還會幫你做這些數據的恢復操作。所以線程的切換非常耗性能。但是協程的切換只是單純的操作CPU的上下文，所以一秒鐘切換個上百萬次系統都抗的住。

Num03–>協程帶來的問題

協程有一個問題，就是系統并不感知，所以操作系統不會幫你做切換。那么誰來幫你做切換？讓需要執行的協程更多的獲得CPU時間才是問題的關鍵。

舉個例子如下：

目前的協程框架一般都是設計成 1:N 模式。所謂 1:N 就是一個線程作為一個容器里面放置多個協程。那么誰來適時的切換這些協程？答案是有協程自己主動讓出CPU，也就是每個協程池里面有一個調度器，這個調度器是被動調度的。意思就是他不會主動調度。而且當一個協程發現自己執行不下去了(比如異步等待網絡的數據回來，但是當前還沒有數據到)，這個時候就可以由這個協程通知調度器，這個時候執行到調度器的代碼，調度器根據事先設計好的調度算法找到當前最需要CPU的協程。切換這個協程的CPU上下文把CPU的運行權交個這個協程，直到這個協程出現執行不下去需要等等的情況，或者它調用主動讓出CPU的API之類，觸發下一次調度。

Num04–>協程的好處

在IO密集型的程序中由于IO操作遠遠慢于CPU的操作，所以往往需要CPU去等IO操作。同步IO下系統需要切換線程，讓操作系統可以在IO過程中執行其他的東西。這樣雖然代碼是符合人類的思維習慣但是由于大量的線程切換帶來了大量的性能的浪費，尤其是IO密集型的程序。

所以人們發明了異步IO。就是當數據到達的時候觸發我的回調。來減少線程切換帶來性能損失。但是這樣的壞處也是很大的，主要的壞處就是操作被 “分片” 了，代碼寫的不是 “一氣呵成” 這種。而是每次來段數據就要判斷數據夠不夠處理哇，夠處理就處理吧，不夠處理就在等等吧。這樣代碼的可讀性很低，其實也不符合人類的習慣。

但是協程可以很好解決這個問題。比如把一個IO操作寫成一個協程。當觸發IO操作的時候就自動讓出CPU給其他協程。要知道協程的切換很輕的。協程通過這種對異步IO的封裝既保留了性能也保證了代碼的容易編寫和可讀性。在高IO密集型的程序下很好。但是高CPU密集型的程序下沒啥好處。

Num05–>yield實現一個簡單協程案例

import timedef A():  while True:    print("----我是A函數---")    yield    time.sleep(0.5)def B(c):  while True:    print("----我是B函數---")    next(c)    time.sleep(0.5)if __name__ == '__main__':  a = A()  B(a)# 結果如下：# ----我是B函數---# ----我是A函數---# ----我是B函數---# ----我是A函數---# ----我是B函數---# ----我是A函數---# ----我是B函數---# ----我是A函數---# ----我是B函數---# ----我是A函數---# ......

Num06–>greenlet版本實現協程案例

#!/usr/bin/env python# -*- coding: utf-8 -*-# @Author : xiaokefrom greenlet import greenletimport timedef test1():  while True:    print("---我是A函數--")    gr2.switch()    time.sleep(0.5)def test2():  while True:    print("---我是B函數--")    gr1.switch()    time.sleep(0.5)def main():  # 切換到gr1中運行  gr1.switch()if __name__ == '__main__':  gr1 = greenlet(test1)  gr2 = greenlet(test2)  main()# 結果如下：# ---我是A函數--# ---我是B函數--# ---我是A函數--# ---我是B函數--# ---我是A函數--# ---我是B函數--# ---我是A函數--# ---我是B函數--# ......

Num07–>gevent實現協程案例

原理：其原理是當一個greenlet遇到IO(指的是input output 輸入輸出，比如網絡、文件操作等)操作時，比如訪問網絡，就自動切換到其他的greenlet，等到IO操作完成，再在適當的時候切換回來繼續執行。

由于IO操作非常耗時，經常使程序處于等待狀態，有了gevent為我們自動切換協程，就保證總有greenlet在運行，而不是等待IO

#!/usr/bin/env python# -*- coding: utf-8 -*-# @Author : xiaokeimport geventdef task1(n):  for i in range(n):    print('----task1-----')    gevent.sleep(1)    # time.sleep(1) # time.sleep沒有讓gevent感知到等待def task2(n):  for i in range(n):    print('----task2-----')    gevent.sleep(1)    # time.sleep(1)def main():  g1 = gevent.spawn(task1, 5)  g2 = gevent.spawn(task2, 5)  g1.join()  g2.join()if __name__ == "__main__":  main()# 結果如下：# ----task1-----# ----task2-----# ----task1-----# ----task2-----# ----task1-----# ----task2-----# ----task1-----# ----task2-----# ----task1-----# ----task2-----

gevent并發下載器

實際代碼里，我們不會用gevent.sleep()去切換協程，而是在執行到IO操作時，gevent自動切換，代碼如下

#!/usr/bin/env python# -*- coding: utf-8 -*-# @Author : xiaokeimport urllib.request # py3import geventfrom gevent import monkey# 猴子補丁，將標準庫的涉及IO操作方法替換成geventmonkey.patch_all() # 協程的任務函數def my_download(url):  print('GET %s' % url)  response = urllib.request.urlopen(url)  data = response.read()  print('下載 %d bytes from %s' % (len(data), url))def main():  g1 = gevent.spawn(my_download, 'http://www.google.cn')  g2 = gevent.spawn(my_download, 'http://www.qq.com')  g3 = gevent.spawn(my_download, 'http://www.baidu.com')  gevent.joinall([g1, g2, g3]) # 等待指定的協程結束if __name__ == "__main__":  main()# 結果如下：# GET http://www.google.cn# GET http://www.qq.com# GET http://www.baidu.com# 下載 102221 bytes from http://www.baidu.com# 下載 52297 bytes from http://www.qq.com# 下載 3213 bytes from http://www.google.cn#從上能夠看到是先獲取baidu的相關信息，然后依次是qq#google，但是收到數據的先后順序不一定與發送順序相同，#這也就體現出了異步，即不確定什么時候會收到數據，順序不一定.

Num08–>gevent版–TCP服務器

#!/usr/bin/env python# -*- coding: utf-8 -*-# @Author : xiaokeimport socketimport geventfrom gevent import monkey#猴子補丁，將標準庫的涉及IO操作方法替換成geventmonkey.patch_all()# 需要為客戶端提供服務def do_service(connect_socket):  while True:    # tcp recv() 只會返回接收到的數據    recv_data = connect_socket.recv(1024)    # if recv_data == b'':    if len(recv_data) == 0:      # 發送方關閉tcp的連接,recv()不會阻塞，而是直接返回''      # print('client %s close' % str(client_addr))      # s.getpeername()  s.getsockname()      print('client %s close' % str(connect_socket.getpeername()))      break    print('recv: %s' % recv_data.decode('gbk'))def main():  listen_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # 設置允許復用地址,當建立連接之后服務器先關閉，設置地址復用  listen_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  my_addr = ('192.168.105.125', 8080)  listen_socket.bind(my_addr)  listen_socket.listen(5) # 設置套接字成監聽,5表示一個己連接隊列長度  print('listening...')  while True:    # 接受連接請求，創建連接套接字，用于客戶端連通信    connect_socket, client_addr = listen_socket.accept() # accept默認會引起阻塞    # 新創建連接用的socket, 客戶端的地址    # print(connect_socket)    print(client_addr)    # 每當來新的客戶端連接，創建協程，由協程和客戶端通信    coroutine_do_service = gevent.spawn(do_service, connect_socket)if __name__ == "__main__":  main()