🍖Python併發編程之協程

引入

咱們知道一個線程同一時間內只能被操做系統分配一個CPU資源, 咱們能夠基於多進程實現併發, 也能夠基於多線程實現併發, CPU正在運行一個任務, 有兩種狀況下會被切去執行其它任務, 一種是該任務發生了阻塞, 另外一是該任務運行時間過長或者被其餘優先級更高的任務奪走CPU, 對於單線程來講, 若是一個單線程下運行了多個任務, 那麼就不可避免的出現I/O操做, 一旦一個任務出現阻塞的狀況, 那麼整個線程將處於阻塞狀態(由於一旦阻塞, CPU資源將會被奪走), 可是若是咱們能在本身的應用程序中(即用戶級別, 非操做系統級別)控制單線程下多個任務能在一個任務遇到I/O以後立馬切換到另外一個任務, 那麼咱們就能夠保證該線程能最大限度的保持就緒狀態(也就是隨時能進入運行態), 至關於咱們在應用程序級別將I/O操做隱藏起來, 迷惑操做系統, 讓其看到的狀態就是一直在計算, I/O比較少, 因而操做系統就會將更多的CPU資源分配給該線程

一. 什麼是協程

1.回顧併發的本質

• CPU在多個任務之間來回切換, 而且在切換以前保存好當前的狀態

2.協程是什麼

• 基於單線程下實現的併發, 又稱爲微線程(Coroutine), 它是一種用戶態的輕量級線程, 即協程是由應用程序本身控制調度的
• 與進程線程同樣, 並非真實存在的東西, 只是程序員爲了方便理解虛擬出來的概念

3.內核級別和用戶級別的調度

• 線程的執行須要被操做系統分配CPU資源, 這屬於內核級別的調度
• 而單個線程內的多個任務之間的調度是由應用程序本身實現的, 這屬於用戶級別的調度

4.內核級別與用戶級別在線程中的調度比較

• 優勢 :
  • 協程的開銷更小, 更輕量,屬於應用程序級別的切換, 操做系統徹底感受不到
  • 單線程下實現併發效果, 更大限度的利用CPU資源 (能夠是一個程序開啓多個進程, 一個進程開啓多個線程, 每一個線程開啓協程)
• 特色 :
  • 本身的應用程序實現的多個任務之間的調度, 遇到I/O就切, 能夠將單線程的I/O降到最低
• 缺點 :
  • 協程是在單線程下實現的, 因此它沒法利用多核
  • 引入協程, 就須要檢測該線程下全部的I/O行爲, 實現遇到I/O就切, 少一個都不行,一旦協程出現了阻塞, 也將會阻塞整個線程

5.使用 yield 模擬協程的效果

• yield 是在生成器那一章學到的知識點, 它能夠保存狀態, 這與操做系統保存線程的狀態很像, 但它是屬於代碼級別控制的, 更輕量

import time

def Foo():
    for i in range(100):
        print(f"--->Foo:{i}")
        yield  # 保存狀態

def Bar():
    f = Foo()
    for i in range(10000000):
        i += 1
        next(f)

start_time = time.time()
Bar()
stop_time = time.time()
print(f"user time:{stop_time-start_time}")

ps : yield沒法檢測I/O, 沒法實現遇到I/O就進行切換

6.總結協程特色

• 在單線程實現的併發
• 修改共享數據不須要加鎖(不會形成同時修改的狀況)
• 用戶程序裏本身保存多個控制流的上下文棧
• 一個協程遇到I/O操做會自動切換到其餘協程

ps : 如何實現自動檢測 I/O, 上面模擬使用的 yield 以及 greenlet 都沒法作到, 因而如下就開始介紹 gevent 模塊(select機制)

二.Gevent模塊介紹

1.安裝Gevent

🍋"cmd" 或 "pycharm" 的 "Terminal"
pip3 install gevent

2.什麼是 gevent 模塊

• Gevent是Python的第三方庫, 它爲各類併發和網絡相關的任務提供了整潔的API, 咱們能夠經過gevent輕鬆實現併發同步或異步編程

• 在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程

• Greenlet所有運行在主程序操做系統進程的內部，但它們被協做式地調度

3.使用方法

• 經常使用方法

方法	做用
gevent.spawn(func,args/kwargs)	建立一個協程對象, 第一個參數是函數名, 後面的參數是函數的位置參數或者關鍵字參數
[協程對象].join( )	等待協程對象的結束
gevent.joinall([對象1,對象2])	等待多個協程對象的結束, 參數是一個列表, 放多個協程對象
[協程對象].value	拿到協程對象的返回值

• 示例:遇到I/O自動切換任務

import gevent
import time

def eat(name):
    print(f"{name}正在吃東西")
    gevent.sleep(3)
    print(f"{name}吃完了")
    return "i am eat"

def play(name):
    print(f"{name}正在玩手機")
    gevent.sleep(1)
    print(f"{name}玩夠了手機")
    return "i am play"

start_time = time.time()
g1 = gevent.spawn(eat,"派大星")
g2 = gevent.spawn(play,"海綿寶寶")

# g1.join()  # 等待協程對象g1結束
# g2.join()  # 等待協程對象g2結束
gevent.joinall([g1,g2])  # 等待協程對象g1和g2結束

print(g1.value)  # 獲取協程對象g1的返回值
print(g2.value)  # 獲取協程對象g2的返回值

print(f"用時:{time.time()-start_time}")

'''輸出
派大星正在吃東西
海綿寶寶正在玩手機
海綿寶寶玩夠了手機
派大星吃完了
i am eat
i am play
用時:3.0330262184143066
'''

4.gevent不支持識別其餘操做的I/O

• 像time.sleep(2) 或者其餘類型的I/O, gevent模塊沒法識別, 只能識別 gevent.sleep(2)

• 解決方法 : 使用猴子補丁讓其能識別 from gevent import monkey;monkey.patch_all(),放在文件開頭

from gevent import monkey;monkey.patch_all()
import gevent
import time
from threading import current_thread

def eat(name):
    print(current_thread().name)  # 查看該線程的名字
    print(f"{name}正在吃東西")
    time.sleep(3)
    print(f"{name}吃完了")
    return "i am eat"

def drink(name):
    print(current_thread().name)  # 查看該線程的名字
    print(f"{name}正在喝湯")
    time.sleep(2)
    print(f"{name}把湯喝完了")
    return "i am drink"

def play(name):
    print(current_thread().name)  # 查看該線程的名字
    print(f"{name}正在玩手機")
    time.sleep(1)
    print(f"{name}玩夠了手機")
    return "i am play"

start_time = time.time()
g1 = gevent.spawn(eat,"派大星")
g2 = gevent.spawn(drink,"章魚哥")
g3 = gevent.spawn(play,"海綿寶寶")

# g1.join()  # 等待協程對象g1結束
# g2.join()  # 等待協程對象g2結束
# g3.join()  # 等待協程對象g3結束
gevent.joinall([g1,g2,g3])  # 等待協程對象g1和g2結束

print(g1.value)  # 獲取協程對象g1的返回值
print(g2.value)  # 獲取協程對象g2的返回值
print(g3.value)  # 獲取協程對象g3的返回值

print(f"用時:{time.time()-start_time}")

'''輸出
Dummy-1
派大星正在吃東西
Dummy-2
章魚哥正在喝湯
Dummy-3
海綿寶寶正在玩手機
海綿寶寶玩夠了手機
章魚哥把湯喝完了
派大星吃完了
i am eat
i am drink
i am play
用時:3.0230190753936768
'''
🍓# 能夠查看到三個線程的名字 : Dummy-一、Dummy-二、Dummy-三、（都是假線程)

三.使用協程編寫socket-TCP程序示例

1.服務端

from gevent import monkey;monkey.patch_all()  # 添加猴子補丁
import gevent
from socket import *

# 建連接循環
def link(ip,port):
    try:
        server = socket(AF_INET, SOCK_STREAM)
        server.bind((ip,port))
        server.listen(5)
    except Exception as E:
        print(E);return
    while 1:
            conn,addr = server.accept()
            gevent.spawn(communication,conn)  
            # 創建連接成功以後開啓一個協程任務進行通訊循環

# 通訊循環
def communication(conn):
    while 1:
        try:
            data = conn.recv(1024)
            if len(data) == 0:break
            conn.send(data.upper())
        except Exception as E:
            print(E);break
    conn.close()

if __name__ == '__main__':
    g1 = gevent.spawn(link,"127.0.0.1",8090)  # 先啓動一個創建連接循環的協程任務
    g1.join()

2.客戶端

• 客戶端的編寫能夠是常規編寫, 而後複製出多臺客戶端

from socket import *

client = socket(AF_INET,SOCK_STREAM)
client.connect(("127.0.0.1",8090))

while 1:
    user = input(">>").strip()
    if len(user) == 0:continue
    client.send(user.encode("utf-8"))
    data = client.recv(1024)
    print(data.decode("utf-8"))

• 也可使用多線程開啓多個客戶端, 不過公用同一個終端屏幕, 效果不明顯

from threading import Thread,current_thread
from socket import *

def connection(ip,port,i):
    client = socket(AF_INET,SOCK_STREAM)
    client.connect((ip,port))
    while 1:
        client.send(f"客戶端編號:{i},名字:{current_thread().name}".encode("utf-8"))
        data = client.recv(1024)
        print(data.decode("utf-8"))

if __name__ == '__main__':
    for i in range(10):  # 多線程開啓 10 個客戶端進行與服務端的鏈接
        t = Thread(target=connection,args=("127.0.0.1",8090,i))
        t.start()