Python的協(xié)程操作幾種實現(xiàn)方式總結

更新時間：2025年09月01日 10:39:46 作者：EvanSun__

在Python中協(xié)程（Coroutine）是一種強大的并發(fā)編程工具,允許你以非阻塞方式處理I/O密集型任務、異步操作等場景,這篇文章主要介紹了Python的協(xié)程操作幾種實現(xiàn)方式,需要的朋友可以參考下

概念

協(xié)程是單線程下的并發(fā)，又稱微線程，纖程。英文名Coroutine。是python個中另外一種實現(xiàn)多任務的方式，只不過比線程更小占用更小執(zhí)行單元（理解為需要的資源）。為啥說它是一個執(zhí)行單元，因為它自帶CPU上下文。這樣只要在合適的時機，我們可以把一個協(xié)程切換到另一個協(xié)程。只要這個過程中保存或恢復 CPU上下文那么程序還是可以運行的。

協(xié)程是一種用戶態(tài)的輕量級線程，即協(xié)程是由用戶程序自己控制調(diào)度的。

通俗理解

在一個線程中的某個函數(shù)，可以在任何地方保存當前函數(shù)的一些臨時變量等信息，然后切換到另外一個函數(shù)中執(zhí)行，注意不是通過調(diào)用函數(shù)的方式做到的，并且切換的次數(shù)以及什么時候再切換到原來的函數(shù)都由開發(fā)者自己確定

協(xié)程和線程差異

在實現(xiàn)多任務時, 線程切換從系統(tǒng)層面遠不止保存和恢復CPU上下文這么簡單。操作系統(tǒng)為了程序運行的高效性每個線程都有自己緩存Cache等等數(shù)據(jù)，操作系統(tǒng)還會幫你做這些數(shù)據(jù)的恢復操作。所以線程的切換非常耗性能。但是協(xié)程的切換只是單純的操作CPU的上下文，所以一秒鐘切換個上百萬次系統(tǒng)都抗的住。

協(xié)程的優(yōu)勢

協(xié)程的切換開銷更小，屬于程序級別的切換，操作系統(tǒng)完全感知不到，因而更加輕量級
單線程內(nèi)就可以實現(xiàn)并發(fā)的效果，最大限度地利用cpu

協(xié)程的缺點

協(xié)程的本質(zhì)是單線程下，無法利用多核，可以是一個程序開啟多個進程，每個進程內(nèi)開啟多個線程，每個線程內(nèi)開啟協(xié)程
協(xié)程指的是單個線程，因而一旦協(xié)程出現(xiàn)阻塞，將會阻塞整個線程

協(xié)程的特點

必須在只有一個單線程里實現(xiàn)并發(fā)
修改共享數(shù)據(jù)不需加鎖
用戶程序里自己保存多個控制流的上下文棧
附加：一個協(xié)程遇到IO操作自動切換到其它協(xié)程（如何實現(xiàn)檢測IO，yield、greenlet都無法實現(xiàn)，就用到了 gevent模塊（select機制））

協(xié)程的本質(zhì)

協(xié)程的本質(zhì)就是在單線程下，由用戶自己控制一個任務遇到io阻塞了就切換另外一個任務去執(zhí)行，以此來提升效率。

實現(xiàn)方式

方式一 Greenlet 模塊

首先要安裝Greenlet模塊 pip install greenlet

import time
from greenlet import greenlet

def study(name):
    print(f"{name} 正在學習面向對象...")
    time.sleep(2)
    g2.switch("qiaoba")		
    print(f"{name} 正在學習協(xié)程...")
    g2.switch("qiaoba")

def play(name):
    print(f"{name} 正在玩超級瑪麗...")
    time.sleep(2)
    g1.switch("lufei")
    print(f"{name} 正在玩CS...")

g1 = greenlet(study)	# xxx = greenlet(fun方法名)
g2 = greenlet(play)
g1.switch("lufei")		# 使用xxx.switch("參數(shù)") 手動切換另一個方法

缺點是遇到IO阻塞不會自動切換，需要人工切換，比較麻煩

方式二 Gevent模塊

python還有一個比greenlet更強大的并且能夠自動切換任務的模塊 gevent

原理是當一個greenlet遇到IO(指的是input output 輸入輸出，比如網(wǎng)絡、文件操作等)操作時，比如訪問網(wǎng)絡，就自動切換到其他的greenlet，等到IO操作完成，再在適當?shù)臅r候切換回來繼續(xù)執(zhí)行。由于IO操作非常耗時，經(jīng)常使程序處于等待狀態(tài)，有了gevent為我們自動切換協(xié)程，就保證總有greenlet在運行，而不是等待IO

首先要安裝Gevent模塊 pip3 install gevent

用法
g1 = gevent.spawn(fun方法名,args參數(shù)1,x=args參數(shù)2...)  # 創(chuàng)建一個協(xié)程對象g1，參數(shù)是傳給方法的位置實參或關鍵字實參
g1.join()  # 等待g1結束
gevent.joinall([g1,g2])  # 等待g1g2全部結束
g1.value   # 獲得g1的返回值

原來的gevent不能識別原程序中的耗時操作的代碼，必須要替換為gevent自帶的方法。

可以通過 打補丁 的方式將程序中用到的耗時操作的代碼，自動替換為gevent中自己實現(xiàn)的模塊，這樣例如time.sleep()等耗時操作gevent便可以識別了

from gevent import monkey
monkey.patch_all()  # 自動替換全部耗時操作的代碼

方式三(模擬) yield

import time

def consumer():
    while 1:
        x = yield 0
        print(x)
        
def producer():
    g = consumer()
    g.__next__()
    for i in range(100):
        g.send(i)
        print(i)
start = time.time()
producer()
print(time.time()-start)  # 0.0010056495666503906

start = time.time()
for i in range(100):
    print(i)
print(time.time()-start)  # 0.0010077953338623047

方法四 asyncio模塊（已過時）

在Python3.4之前官方未提供協(xié)程的類庫，一般大家都是使用greenlet等其他來實現(xiàn)。在Python3.4發(fā)布后官方正式支持協(xié)程，即：asyncio模塊。

import asyncio
@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)  # 遇到IO耗時操作，自動化切換到tasks中的其他任務
print(2)
@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2) # 遇到IO耗時操作，自動化切換到tasks中的其他任務
    print(4)
tasks = [
    asyncio.ensure_future( func1() ),
    asyncio.ensure_future( func2() )
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

方法五 async & awit 模塊

async & awit 關鍵字在Python3.5版本中正式引入，基于他編寫的協(xié)程代碼其實就是上一示例的加強版，讓代碼可以更加簡便。Python3.8之后@asyncio.coroutine 裝飾器就會被移除，推薦使用 async & awit 關鍵字實現(xiàn)協(xié)程代碼。

import asyncio
async def func1():
   print(1)
   await asyncio.sleep(2)
   print(2)
async def func2():
   print(3)
   await asyncio.sleep(2)
   print(4)
tasks = [
   asyncio.ensure_future(func1()),
   asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))