Python协程实战指南：工作中的常见使用模式

在掌握了线程和进程的基础知识后，协程的学习确实会变得相对简单。但是，要在实际项目中熟练运用协程，还需要了解各种常见的使用模式和最佳实践。

基于我在智能体开发和测试自动化项目中的实际经验，我总结了一些最常用、最实用的协程使用方式。这些模式在日常开发中出现频率很高，掌握它们能让你的异步编程更加得心应手。

如果只从工作上理解，那么协程的使用要学会处理一下几个问题

1. 一个简单的协程函数如何写，怎么执行
2. 协程间如何互相调用及获取其结果值
3. 多协程 怎么一起执行，怎么控制并发
4. 协程间的共享变量如何同步
5. 协程如果有异常，就会停止下来吗

那么我一一说明一下上面的问题

一个简单的协程函数如何写，怎么执行

import asyncio


async def test(num):
    print(f"test:{num}")


if __name__ == '__main__':
    asyncio.run(test(1))

输出

test:1

协程间如何互相调用及获取其结果值

import asyncio


async def test(num):
    print(f"test:{num}")
    return "test",num

async def test2():
    resp = await test(1)
    print(resp)
    return resp


if __name__ == '__main__':
    asyncio.run(test2())

[out]

test:1
('test', 1)

多协程怎么控制并发

import asyncio


async def test(num):
    print(f"test:{num}")
    return "test",num

async def test2():
    tasks = [test(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)
    return results




if __name__ == '__main__':
    asyncio.run(test2())

【out】

test:0
test:1
test:2
test:3
test:4
test:5
test:6
test:7
test:8
test:9
('test', 0)
('test', 1)
('test', 2)
('test', 3)
('test', 4)
('test', 5)
('test', 6)
('test', 7)
('test', 8)
('test', 9)

并发控制

import asyncio


async def test(num):
    print(f"test:{num}")
    return "test",num

async def test2():
    concurrency_limit =2
    semaphore = asyncio.Semaphore(concurrency_limit)  # 创建信号量
    tasks = []
    for i in range(10):
        tasks.append(test(i))
    async with semaphore:
        results = await asyncio.gather(*tasks)
    print(results)
    for result in results:
        print(result)
    return results


if __name__ == '__main__':
    asyncio.run(test2())

把并发写成一个函数

    async def switch_group_and_exec_edl_cmd(self, group_ids, concurrency_limit=3):
        semaphore = asyncio.Semaphore(concurrency_limit)  # 创建信号量

        async def limited_exec(usb_id):
            async with semaphore:  # 限制并发
                await self.test_edl_on_off(usb_id)

        for group_id in group_ids:
            start_time = time.time()
            tasks = []
            for usb_id in self.group_usb_ids.get(group_id, []):
                tasks.append(limited_exec(usb_id))  # 使用信号量限制并发
            await asyncio.gather(*tasks)  # 执行所有任务

协程间的共享变量如何同步

import asyncio

class SharedResource:
    def __init__(self):
        self.value = 0
        self.lock = asyncio.Lock()  # 创建锁

    async def increment(self):
        async with self.lock:  # 获取锁
            self.value += 1
            print(f"Value incremented to: {self.value}")

async def worker(shared):
    for _ in range(5):
        await shared.increment()

async def main():
    shared = SharedResource()
    tasks = [worker(shared) for _ in range(3)]
    await asyncio.gather(*tasks)


if __name__ == '__main__':
    asyncio.run(main())

协程如果有异常，就会停止下来吗

这点和正常的使用，没有不同

import asyncio
import random

async def test(num):
    print(f"test:{num}")
    # 随机出现异常，停止
    if random.random() < 0.5:
        raise Exception("test")
    return "test",num

async def test2():
    concurrency_limit =2
    semaphore = asyncio.Semaphore(concurrency_limit)  # 创建信号量
    tasks = []
    for i in range(10):
        tasks.append(test(i))
    async with semaphore:
        results = await asyncio.gather(*tasks)
    print(results)
    for result in results:
        print(result)
    return results


if __name__ == '__main__':
    asyncio.run(test2())

异步迭代器使用

import asyncio

class AsyncIterator:
    def __init__(self, limit):
        self.limit = limit
        self.count = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.count < self.limit:
            await asyncio.sleep(1)  # 模拟异步操作
            self.count += 1
            return self.count
        else:
            raise StopAsyncIteration

async def main():
    # 方式一
    # async for i in AsyncIterator(5):
    #     print(i)
    # 方式二
    list_data =  AsyncIterator(5)
    print(list_data)
    async for i in list_data:
        print(i)

# 运行异步主函数
asyncio.run(main())

import asyncio

# 定义一个异步生成器
async def async_generator():
    for i in range(5):
        await asyncio.sleep(1)  # 模拟异步操作
        yield i  # 生成一个值

# 使用 async for 迭代异步生成器
async def main():
    async for value in async_generator():
        print(value)

# 运行主协程
asyncio.run(main())