Python協(xié)程有哪些使用技巧

小樊
81
2024-10-30 20:37:37

Python協(xié)程（Coroutine）是一種輕量級(jí)的線(xiàn)程，它可以在執(zhí)行過(guò)程中掛起并在稍后恢復(fù)。協(xié)程在異步編程和并發(fā)處理中非常有用。以下是一些使用Python協(xié)程的技巧：

  1. 使用async def定義協(xié)程函數(shù)：協(xié)程函數(shù)使用async def關(guān)鍵字定義，而不是普通的def。
async def my_coroutine():
    print("Hello, coroutine!")
  1. 使用await關(guān)鍵字調(diào)用協(xié)程：在協(xié)程函數(shù)內(nèi)部，使用await關(guān)鍵字調(diào)用其他協(xié)程函數(shù)或異步操作。這會(huì)讓當(dāng)前協(xié)程掛起，直到被調(diào)用的協(xié)程完成。
async def main():
    await my_coroutine()
  1. 使用asyncio.run()啟動(dòng)協(xié)程：asyncio.run()函數(shù)用于運(yùn)行頂層的協(xié)程，并等待其完成。這是啟動(dòng)協(xié)程的推薦方式。
import asyncio

async def main():
    await my_coroutine()

asyncio.run(main())
  1. 使用asyncio.gather()并發(fā)運(yùn)行多個(gè)協(xié)程：asyncio.gather()函數(shù)接受一個(gè)協(xié)程列表，并并發(fā)運(yùn)行它們。當(dāng)所有協(xié)程完成時(shí)，它返回一個(gè)包含所有協(xié)程結(jié)果的列表。
import asyncio

async def my_coroutine(n):
    await asyncio.sleep(n)
    return n

async def main():
    coroutines = [my_coroutine(i) for i in range(5)]
    results = await asyncio.gather(*coroutines)
    print(results)

asyncio.run(main())
  1. 使用asyncio.Queue()進(jìn)行協(xié)程間通信：asyncio.Queue()類(lèi)用于在協(xié)程之間傳遞數(shù)據(jù)。生產(chǎn)者協(xié)程將數(shù)據(jù)放入隊(duì)列，消費(fèi)者協(xié)程從隊(duì)列中取出數(shù)據(jù)。
import asyncio

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(1)

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    prod_task = asyncio.create_task(producer(queue))
    cons_task = asyncio.create_task(consumer(queue))

    await prod_task
    queue.join()

    cons_task.cancel()
    try:
        await cons_task
    except asyncio.CancelledError:
        pass

asyncio.run(main())
  1. 使用asyncio.Semaphore()限制并發(fā)協(xié)程數(shù)量：asyncio.Semaphore()類(lèi)用于限制同時(shí)運(yùn)行的協(xié)程數(shù)量。協(xié)程在嘗試獲取信號(hào)量時(shí)會(huì)被掛起，直到信號(hào)量可用。
import asyncio

async def my_coroutine(semaphore, n):
    async with semaphore:
        print(f"Coroutine {n} started")
        await asyncio.sleep(1)
        print(f"Coroutine {n} finished")

async def main():
    semaphore = asyncio.Semaphore(3)
    coroutines = [my_coroutine(semaphore, i) for i in range(10)]
    await asyncio.gather(*coroutines)

asyncio.run(main())

這些技巧可以幫助你更有效地使用Python協(xié)程進(jìn)行異步編程和并發(fā)處理。

0 踩