asyncio

asyncio 模块提供异步编程功能，用于非阻塞式并发操作。

适合处理多个并发任务，如传感器读取、网络请求、串口通信等
使用 async/await 语法
避免使用 time.sleep() 导致的阻塞

基本用法

import asyncio

# 定义异步函数
async def hello():
    print('Hello')
    await asyncio.sleep(1)  # 非阻塞延时
    print('World')

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

核心概念

协程（Coroutine）：使用 async def 定义的函数
任务（Task）：协程的执行实例
事件循环（Event Loop）：管理协程的执行

常用函数

1. 运行协程

# 运行单个协程
asyncio.run(coro())

# 获取事件循环
loop = asyncio.get_event_loop()
# 运行协程直到完成
loop.run_until_complete(coro())
# 启动事件循环（持续运行）
loop.run_forever()

2. 延时函数

# 非阻塞延时（替代 time.sleep）
await asyncio.sleep(1)  # 延时 1 秒

3. 并发执行

# 同时运行多个协程
await asyncio.gather(
    coro1(),
    coro2(),
    coro3()
)

# 创建任务
async def main():
    task1 = asyncio.create_task(coro1())
    task2 = asyncio.create_task(coro2())

    await task1
    await task2

asyncio.run(main())

示例1：异步 LED 闪烁

from machine import Pin
import asyncio

led = Pin(25, Pin.OUT)

async def blink():
    while True:
        led.on()
        await asyncio.sleep(0.5)  # 非阻塞延时
        led.off()
        await asyncio.sleep(0.5)

# 运行
asyncio.run(blink())

示例2：多个任务并发

from machine import Pin
import asyncio

led1 = Pin(25, Pin.OUT)
led2 = Pin(15, Pin.OUT)

async def blink1():
    while True:
        led1.on()
        await asyncio.sleep(0.5)
        led1.off()
        await asyncio.sleep(0.5)

async def blink2():
    while True:
        led2.on()
        await asyncio.sleep(1)
        led2.off()
        await asyncio.sleep(1)

async def main():
    # 并发运行两个任务
    await asyncio.gather(
        blink1(),
        blink2()
    )

asyncio.run(main())

示例3：异步传感器读取

from machine import ADC, Pin
import asyncio

adc = ADC(Pin(26))

async def read_sensor():
    while True:
        value = adc.read_u16()
        voltage = value * 3.3 / 65535
        print(f'传感器值: {value}, 电压: {voltage:.2f}V')
        await asyncio.sleep(1)  # 每秒读取一次

async def other_task():
    while True:
        print('其他任务运行中...')
        await asyncio.sleep(2)  # 每 2 秒执行一次

async def main():
    await asyncio.gather(
        read_sensor(),
        other_task()
    )

asyncio.run(main())

示例4：异步串口通信

from machine import UART, Pin
import asyncio

uart = UART(0, baudrate=115200, tx=Pin(0), rx=Pin(1))

async def read_uart():
    while True:
        if uart.any():
            data = uart.read()
            if data:
                print('收到:', data.decode('utf-8'))
        await asyncio.sleep(0.1)  # 100ms 检查一次

async def write_uart():
    while True:
        message = 'Hello from async\r\n'
        uart.write(message)
        print('发送:', message.strip())
        await asyncio.sleep(3)  # 每 3 秒发送一次

async def main():
    await asyncio.gather(
        read_uart(),
        write_uart()
    )

asyncio.run(main())

示例5：超时处理

import asyncio

async def long_task():
    await asyncio.sleep(5)  # 模拟长时间任务
    return '完成'

async def main():
    try:
        # 设置 2 秒超时
        result = await asyncio.wait_for(long_task(), timeout=2)
        print('结果:', result)
    except asyncio.TimeoutError:
        print('任务超时')

asyncio.run(main())

注意事项

MicroPython 支持：并非所有 MicroPython 固件都包含 asyncio，需要确认固件是否支持
性能：异步编程适合 I/O 密集型任务，不适合 CPU 密集型任务
语法：
- 使用 async def 定义协程
- 使用 await 等待协程完成
- 不能在同步函数中使用 await
事件循环：
- 每个线程只能有一个事件循环
- asyncio.run() 会创建并运行一个新的事件循环
阻塞操作：
- 避免在协程中使用阻塞操作（如 time.sleep()）
- 应使用 await asyncio.sleep() 代替

常见问题

Q: 为什么我的协程没有执行？ A: 协程定义后不会自动执行，需要通过 asyncio.run() 或事件循环来运行。

Q: 如何取消一个任务？ A: 使用 task.cancel() 取消任务，任务会抛出 asyncio.CancelledError 异常。

Q: 异步编程和多线程的区别？ A: 异步编程是单线程的，通过事件循环实现并发；多线程是真正的并行执行。