快捷導(dǎo)航

Python中協(xié)程間通信的方式小結(jié)

更新時間：2025年01月09日 14:46:08 作者：Toormi

Python中協(xié)程間的通信方式包括asyncio.Queue、asyncio.Event、asyncio.Condition、asyncio.Semaphore、asyncio.Streams和asyncio.Future,感興趣的可以了解一下

1. asyncio.Queue

asyncio.Queue 是 Python asyncio 庫提供的一個線程安全的隊列，用于在協(xié)程之間傳遞數(shù)據(jù)。與常規(guī)的隊列類似，但它支持異步操作，即可以在協(xié)程內(nèi)等待隊列中的數(shù)據(jù)。

示例：

import asyncio

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        print(f"Produced: {i}")
        await asyncio.sleep(1)

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:  # 用 None 表示終止信號
            break
        print(f"Consumed: {item}")
        await asyncio.sleep(2)

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(producer(queue), consumer(queue))

asyncio.run(main())

2. asyncio.Event

asyncio.Event 是一個簡單的同步原語，允許一個協(xié)程等待某個事件的發(fā)生，另一個協(xié)程則負(fù)責(zé)設(shè)置該事件。通常用于通知其他協(xié)程某些狀態(tài)發(fā)生變化。

示例：

import asyncio

async def waiter(event):
    print("Waiting for event...")
    await event.wait()  # 阻塞直到事件被觸發(fā)
    print("Event occurred!")

async def setter(event):
    await asyncio.sleep(2)
    print("Setting event!")
    event.set()  # 設(shè)置事件

async def main():
    event = asyncio.Event()
    await asyncio.gather(waiter(event), setter(event))

asyncio.run(main())

3. asyncio.Condition

asyncio.Condition 類似于線程中的 Condition，允許一個或多個協(xié)程等待某個條件的變化。它可以與鎖結(jié)合使用。

示例：

import asyncio

async def consumer(condition, shared_data):
    async with condition:
        while shared_data["item"] is None:
            await condition.wait()  # 等待條件
        print(f"Consumed: {shared_data['item']}")
        shared_data["item"] = None

async def producer(condition, shared_data):
    await asyncio.sleep(1)
    async with condition:
        shared_data["item"] = "Apple"
        condition.notify()  # 通知等待的協(xié)程
        print(f"Produced: {shared_data['item']}")

async def main():
    shared_data = {"item": None}
    condition = asyncio.Condition()
    await asyncio.gather(producer(condition, shared_data), consumer(condition, shared_data))

asyncio.run(main())

4. asyncio.Semaphore

asyncio.Semaphore 是一個計數(shù)信號量，控制并發(fā)訪問的協(xié)程數(shù)目。適用于限制協(xié)程的并發(fā)數(shù)量。

示例：

import asyncio

async def task(sem):
    async with sem:
        print("Task started")
        await asyncio.sleep(1)
        print("Task completed")

async def main():
    sem = asyncio.Semaphore(2)  # 最多同時運行2個協(xié)程
    tasks = [task(sem) for _ in range(5)]
    await asyncio.gather(*tasks)

asyncio.run(main())

5. asyncio.Streams (StreamReader 和 StreamWriter)

StreamReader 和 StreamWriter 是用于網(wǎng)絡(luò)通信的流接口，適用于兩個協(xié)程之間通過網(wǎng)絡(luò)協(xié)議傳輸數(shù)據(jù)的場景。盡管它們主要用于處理網(wǎng)絡(luò) I/O，但也可以用于在協(xié)程之間傳輸數(shù)據(jù)。

示例：

import asyncio

async def echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print(f"Received {message} from {addr}")

    print("Send: %r" % message)
    writer.write(data)
    await writer.drain()

    print("Closing the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        echo, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

6. asyncio.Future

asyncio.Future 類似于 Promise，它表示一個尚未完成的操作。一個協(xié)程可以將其結(jié)果存儲在 Future 對象中，而其他協(xié)程可以等待該 Future 對象完成。

示例：

import asyncio

async def set_future(future):
    await asyncio.sleep(1)
    future.set_result('Hello from Future!')

async def get_future(future):
    result = await future
    print(f"Got result: {result}")

async def main():
    future = asyncio.Future()
    await asyncio.gather(set_future(future), get_future(future))

asyncio.run(main())

這些通信方式各有其特定的用途，可以根據(jù)不同的場景選擇合適的方式來進(jìn)行協(xié)程間的通信。

到此這篇關(guān)于Python中協(xié)程間通信的方式小結(jié)的文章就介紹到這了,更多相關(guān)Python 協(xié)程間通信內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: