树莓派上如何用Python的Bleak库实现多BLE连接？

核心思路 #

要实现多设备并发通信，核心是利用asyncio的并发机制，将单个设备的通信逻辑封装为独立异步任务，同时替换阻塞式的input为异步输入处理，避免卡住事件循环。

重构后的完整代码示例 #

import asyncio
from bleak import BleakClient

# 替换为你的实际UUID
UUID_RX = "0000ffb2-0000-1000-8000-00805f9b34fb"
UUID_TX = "0000ffb1-0000-1000-8000-00805f9b34fb"

def rx_data_received(device_addr, data):
    """带设备地址的接收回调，区分不同设备的数据"""
    print(f"[{device_addr}] 收到数据: {data.hex()}")

async def handle_device(address, active_clients):
    """单个设备的连接与通信逻辑"""
    async with BleakClient(address) as client:
        active_clients[address] = client
        print(f"已连接设备: {address}")
        await client.start_notify(UUID_RX, lambda _, data: rx_data_received(address, data))
        
        # 保持连接直到主动断开或设备离线
        while client.is_connected:
            await asyncio.sleep(1)
        
        del active_clients[address]
        print(f"设备断开连接: {address}")

async def handle_user_input(active_clients):
    """异步处理用户输入，避免阻塞事件循环"""
    while True:
        # 用to_thread包装同步input，不阻塞异步任务
        raw_input = await asyncio.to_thread(
            input, 
            "输入命令(格式: [设备地址] [指令], q退出): "
        )
        raw_input = raw_input.strip()
        
        if raw_input == "q":
            # 退出前断开所有设备
            for addr, client in active_clients.items():
                if client.is_connected:
                    await client.disconnect()
            print("正在退出...")
            break
        
        # 解析命令
        parts = raw_input.split(maxsplit=1)
        if len(parts) != 2:
            print("格式错误，请输入: 设备地址 + 空格 + 指令")
            continue
        
        target_addr, cmd = parts
        if target_addr not in active_clients:
            print(f"未找到设备: {target_addr}")
            continue
        
        client = active_clients[target_addr]
        if client.is_connected:
            # 发送指令（限制前20字节，和原逻辑一致）
            await client.write_gatt_char(UUID_TX, bytearray(cmd[:20]), True)
            print(f"已发送指令到[{target_addr}]: {cmd}")
        else:
            print(f"设备[{target_addr}]已断开")

if __name__ == "__main__":
    # 要连接的多设备地址列表
    DEVICE_LIST = [
        "AA:BB:CC:DD:EE:FF",
        "11:22:33:44:55:66"
    ]
    
    # 管理活跃连接的客户端字典
    active_clients = {}
    loop = asyncio.get_event_loop()
    
    # 创建所有设备的异步任务
    device_tasks = [
        asyncio.create_task(handle_device(addr, active_clients)) 
        for addr in DEVICE_LIST
    ]
    # 创建输入处理任务
    input_task = asyncio.create_task(handle_user_input(active_clients))
    
    # 并发运行所有任务
    try:
        loop.run_until_complete(asyncio.gather(*device_tasks, input_task))
    except KeyboardInterrupt:
        print("\n收到中断信号，正在退出...")
        for client in active_clients.values():
            if client.is_connected:
                loop.run_until_complete(client.disconnect())

关键细节说明 #

• 并发任务管理：用asyncio.gather同时启动所有设备的连接任务和输入处理任务，实现多设备并行通信。
• 异步输入处理：通过asyncio.to_thread将同步的input放到单独线程执行，避免阻塞事件循环导致其他设备任务暂停。
• 设备状态追踪：用字典active_clients维护当前连接的设备，方便输入命令时快速定位目标设备。
• 回调区分设备：修改接收回调函数，带上设备地址参数，能明确区分不同设备发来的数据。

扩展建议 #

• 如果不同设备使用不同的UUID，可以将UUID作为参数传入handle_device函数。
• 如需自动重连，可在handle_device中添加循环，断开后尝试重新连接。
• 树莓派BLE适配器的最大连接数有限（通常5-7个），不要超过硬件限制。

内容的提问来源于stack exchange，提问作者Michael