如何在Kivy应用中用async_runTouchApp实现蓝牙异步连接？

核心问题解决思路 #

之前用Thread+asyncio.run启动蓝牙任务，找不到设备时没法优雅终止线程和异步任务。改用Kivy官方支持的async_runTouchApp，可以让UI事件和异步蓝牙任务共用同一个事件循环，既避免线程安全问题，又能在设备未找到时直接终止任务，无需额外处理线程。

步骤1：修改应用入口（main.py） #

把常规的runTouchApp替换成异步启动方式，确保整个应用运行在asyncio事件循环中：

from kivy.app import App
from kivy.lang import Builder
from kivy.asyncio import async_runTouchApp
import asyncio

# 导入你的自定义屏幕和蓝牙类
from first_screen import FirstScreen
from bluetooth import Bluetooth_board

# 加载kv文件（根据你的项目结构调整）
Builder.load_file('first_screen.kv')

class MainApp(App):
    def build(self):
        self.bluetooth = Bluetooth_board()
        return FirstScreen(bluetooth=self.bluetooth)

async def main():
    await MainApp().async_run(async_lib="asyncio")

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

步骤2：重构Bluetooth_board类（bluetooth.py） #

去掉线程相关代码，直接用异步任务处理蓝牙连接，设备未找到时直接终止任务：

import asyncio
from kivy.properties import StringProperty
from kivy.event import EventDispatcher
import bleak

device_name = "my_board" # 替换为你的蓝牙设备名称/地址
SwitchCharacteristic = "beb5483e-36e1-4688-b7f5-ea07361b26a8"

class Bluetooth_board(EventDispatcher):
    status = StringProperty("Conectar")
    client = None
    notify_task = None  # 保存通知任务，方便后续取消

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
    
    def run_bleak(self):
        # 按钮点击触发的同步方法，内部创建异步任务
        if self.client and self.client.is_connected:
            print("已处于连接状态")
            return
        # 用create_task把异步任务丢到事件循环，不阻塞UI
        asyncio.create_task(self.board())

    def callback(self, sender: bleak.BleakGATTCharacteristic, data: bytearray):
        text = data.decode(encoding="utf-8")
        print(text)

    async def board(self):    
        self.status = "连接中..."
        devices = await bleak.BleakScanner.discover()
        
        # 过滤目标设备
        target_device = None
        for device in devices:
            if device.name == device_name:
                target_device = device
                break
        
        if not target_device:
            print("未找到目标设备")
            self.status = "未找到设备"
            return  # 直接返回，终止当前异步任务
        
        print(f"找到设备: {target_device.name} - {target_device.address}")
        self.client = bleak.BleakClient(target_device)

        try:
            await self.client.connect()
            self.status = "已连接"
            print("等待数据...")
            
            # 启动通知任务，bleak会持续触发回调直到任务取消
            self.notify_task = asyncio.create_task(
                self.client.start_notify(SwitchCharacteristic, self.callback)
            )
            await self.notify_task
        except Exception as e:
            print(f"连接出错: {e}")
            self.status = "连接失败"
        finally:
            # 清理资源
            if self.client:
                await self.client.disconnect()
                self.client = None
            if self.notify_task:
                self.notify_task.cancel()

关键修改说明 #

1. 移除线程依赖：不再用Thread启动异步任务，改用asyncio.create_task直接在Kivy的事件循环中调度，避免线程间状态同步问题。
2. 优雅终止任务：设备未找到时直接return，异步任务自动结束，无需手动终止线程。
3. 资源清理：新增notify_task保存通知任务，在finally块中取消并断开连接，避免资源泄漏。
4. 统一事件循环：用async_runTouchApp启动应用，让Kivy的UI事件和蓝牙异步任务共用一个循环，确保状态更新的线程安全。

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