使用Bleak连接多台BLE设备时嵌套with语句混乱，求可扩展实现方案

核心思路 #

• 使用contextlib.AsyncExitStack批量管理多个异步上下文，替代嵌套async with语法，可动态适配任意数量的BLE设备
• 将设备地址与对应回调配对为配置项，新增设备仅需更新配置列表，无需修改核心逻辑
• 支持串行/并行两种连接模式，可根据需求选择

可扩展实现代码 #

import asyncio
import bleak
from bleak import BleakClient
from contextlib import AsyncExitStack

# 设备配置：格式为(设备MAC地址, 对应通知回调函数)，新增设备仅需在此添加条目
DEVICE_CONFIGS = [
    ("AA:BB:CC:DD:EE:01", callback0),
    ("AA:BB:CC:DD:EE:02", callback1),
    ("AA:BB:CC:DD:EE:03", callback2),
    ("AA:BB:CC:DD:EE:04", callback3),
    # 继续添加更多设备配置即可
]
UUID = "你的特征UUID"
controllers = []
MAX_RETRY = 3  # 避免无限递归设置最大重试次数

async def connect_single_device(exit_stack: AsyncExitStack, address: str, callback):
    """单设备连接逻辑，可单独调用支持并行连接"""
    client = await exit_stack.enter_async_context(BleakClient(address))
    await client.start_notify(UUID, callback)
    print(f"Connected: {client.is_connected}，地址：{address}")
    controllers.append(client)
    return client

async def main(retry_count=0):
    try:
        async with AsyncExitStack() as exit_stack:
            # 串行连接版本：设备数量少可直接用
            for address, callback in DEVICE_CONFIGS:
                await connect_single_device(exit_stack, address, callback)
            
            # 并行连接版本：设备数量多建议用，大幅缩短连接耗时，注释掉上面串行版打开下面即可
            # connect_tasks = [connect_single_device(exit_stack, addr, cb) for addr, cb in DEVICE_CONFIGS]
            # await asyncio.gather(*connect_tasks)

            # [原业务逻辑代码]

            # 统一停止所有通知
            for client in controllers:
                await client.stop_notify(UUID)
    except bleak.exc.BleakDBusError:
        if retry_count < MAX_RETRY:
            print(f"连接出错，第{retry_count+1}次重试")
            await main(retry_count+1)
        else:
            print("达到最大重试次数，连接失败")

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

补充说明 #

• 新增设备仅需在DEVICE_CONFIGS中添加对应地址和回调即可，无需修改连接逻辑
• AsyncExitStack会在上下文退出时自动断开所有已连接的BLE设备，无需手动处理关闭逻辑
• 原有递归重连逻辑增加了重试次数限制，避免异常情况下无限递归导致栈溢出

内容的提问来源于stack exchange，提问作者Anders Haarberg Eriksen