【python asyncio模块的协程使用示例】_asyncio实例集锦

asyncio是Python的一个异步I/O框架，允许使用async/await语法进行并发编程。以下是多个使用asyncio的例子：

异步执行的基本示例

import asyncio

async def hello_world():
    print("Hello, World!")

asyncio.run(hello_world())
1
2
3
4
5
6

异步休眠

import asyncio

async def async_sleep(duration):
    print(f"Sleeping for {duration} seconds...")
    await asyncio.sleep(duration)
    print(f"Slept for {duration} seconds")

asyncio.run(async_sleep(3))
1
2
3
4
5
6
7
8

异步获取网页内容

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

url = "https://www.example.com/"
html = asyncio.run(fetch(url))
print(html)
1
2
3
4
5
6
7
8
9
10
11

多个异步任务并发执行

import asyncio

async def print_numbers():
    for i in range(10):
        print(i)
        await asyncio.sleep(1)

async def print_letters():
    for letter in 'abcdefghij':
        print(letter)
        await asyncio.sleep(2)

async def main():
    task1 = asyncio.create_task(print_numbers())
    task2 = asyncio.create_task(print_letters())

    await asyncio.gather(task1, task2)

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

使用异步队列进行生产者-消费者模式

import asyncio
import random

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

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

async def main():
    queue = asyncio.Queue()
    producer_task = asyncio.create_task(producer(queue))
    consumer_task = asyncio.create_task(consumer(queue))

    await producer_task
    await queue.put(None)
    await consumer_task

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

以上示例展示了使用asyncio执行基本的异步任务，异步休眠，获取网页内容，同时执行多个任务和实现生产者-消费者模式。

异步执行多个协程并获取结果

import asyncio

async def square(x):
    await asyncio.sleep(1)
    return x * x

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

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12

异步执行有超时限制的任务

pythonCopy codeimport asyncio

async def long_running_task():
    await asyncio.sleep(5)
    return "Task completed"

async def main():
    try:
        result = await asyncio.wait_for(long_running_task(), timeout=3)
    except asyncio.TimeoutError:
        print("Task timeout")
    else:
        print(result)

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

异步执行周期性任务

pythonCopy codeimport asyncio

async def periodic_task(interval):
    while True:
        print("Periodic task executed")
        await asyncio.sleep(interval)

async def main():
    task = asyncio.create_task(periodic_task(2))
    await asyncio.sleep(10)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Periodic task cancelled")

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17