快捷導(dǎo)航

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

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

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

概念

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

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

通俗理解

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

協(xié)程和線程差異

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

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

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

協(xié)程的缺點(diǎn)

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

協(xié)程的特點(diǎn)

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

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

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

實(shí)現(xiàn)方式

方式一 Greenlet 模塊

首先要安裝Greenlet模塊 pip install greenlet

import time
from greenlet import greenlet

def study(name):
    print(f"{name} 正在學(xué)習(xí)面向?qū)ο?..")
    time.sleep(2)
    g2.switch("qiaoba")		
    print(f"{name} 正在學(xué)習(xí)協(xié)程...")
    g2.switch("qiaoba")

def play(name):
    print(f"{name} 正在玩超級(jí)瑪麗...")
    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ù)") 手動(dòng)切換另一個(gè)方法

缺點(diǎn)是遇到IO阻塞不會(huì)自動(dòng)切換，需要人工切換，比較麻煩

方式二 Gevent模塊

python還有一個(gè)比greenlet更強(qiáng)大的并且能夠自動(dòng)切換任務(wù)的模塊 gevent

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

首先要安裝Gevent模塊 pip3 install gevent

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

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

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

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

方式三(模擬) 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模塊（已過(guò)時(shí)）

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

import asyncio
@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)  # 遇到IO耗時(shí)操作，自動(dòng)化切換到tasks中的其他任務(wù)
print(2)
@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2) # 遇到IO耗時(shí)操作，自動(dòng)化切換到tasks中的其他任務(wù)
    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 關(guān)鍵字在Python3.5版本中正式引入，基于他編寫(xiě)的協(xié)程代碼其實(shí)就是上一示例的加強(qiáng)版，讓代碼可以更加簡(jiǎn)便。Python3.8之后@asyncio.coroutine 裝飾器就會(huì)被移除，推薦使用 async & awit 關(guān)鍵字實(shí)現(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))