如何在Kivy中终止while循环？解决串口采集应用关闭卡顿问题

问题背景 #

开发了从Arduino到Python的数据记录仪，用Kivy展示数据，使用while循环后关闭应用时卡顿，原因是后台循环线程未终止，尝试添加终止按钮无效。

解决方案一：线程+可外部控制的终止标志 #

原代码核心问题：

• 循环的flag是局部变量，外部无法修改终止循环
• 每次循环重复打开串口，效率低且易引发资源占用
• 未实现关闭应用时的线程终止逻辑

修改后的Python代码：

import serial
from kivy.app import App
from kivy.lang import Builder
from kivy.uix.screenmanager import Screen, ScreenManager
from kivy.properties import StringProperty
from kivy.clock import mainthread
import threading
import time

class NaviWindow(Screen):
    temp_text = StringProperty("")
    humid_text = StringProperty("")

    def __init__(self, **kwargs):
        super(NaviWindow, self).__init__(**kwargs)
        self.running_flag = True  # 改为实例变量，外部可访问
        self.ser = None  # 提前声明串口对象
        threading.Thread(target=self.tmhm, daemon=True).start()  # 设置为守护线程

    def tmhm(self):
        # 只初始化一次串口，避免重复打开关闭
        try:
            self.ser = serial.Serial('COM3', 115200)
            while self.running_flag:
                data = str(self.ser.readline().strip().decode())
                self.update_values(data.split(","))
                time.sleep(60)
        except Exception as e:
            print(f"串口读取错误: {e}")
        finally:
            if self.ser and self.ser.is_open:
                self.ser.close()  # 确保关闭串口

    @mainthread
    def update_values(self, data):
        if len(data) >=2:  # 增加数据合法性判断
            self.temp_text = data[0]
            self.humid_text = data[1]

    def stop_recording(self):
        self.running_flag = False  # 修改标志终止循环
        if self.ser and self.ser.is_open:
            self.ser.close()

class WindowManager(ScreenManager):
    pass

kys = Builder.load_file("temp4.kv")

class MyMainApp(App):
    def build(self):
        return kys

    def close_application(self):
        # 获取当前屏幕，调用停止方法
        current_screen = self.root.current_screen
        if isinstance(current_screen, NaviWindow):
            current_screen.stop_recording()
        self.stop()  # 关闭Kivy应用

if __name__ == "__main__":
    MyMainApp().run()

对应的KV代码（修正布局结构）：

WindowManager:
    NaviWindow:

<NaviWindow>:
    orientation:'vertical'
    BoxLayout:
        orientation: 'vertical'
        Label:
            text: "Temperature"
            font_size: 30
            size_hint: (1, 0.17)
        Label:
            text: "every 1 minutes"
            font_size: 15
            size_hint: (1, 0.17)
        Label:
            text: "Humidity"
            font_size: 30
            size_hint: (1, 0.17)
        Label:
            text: "every 1 minutes"
            font_size: 15
            size_hint: (1, 0.17)
        Label:
            text: root.temp_text
            font_size: 20
            size_hint: (1, 0.17)
        Label:
            text: root.humid_text
            font_size: 20
            size_hint: (1, 0.17)
        Button:
            size_hint: (1, 0.17)
            text:"Close App"
            on_release: app.close_application()

关键改动说明：

• 将flag改为实例变量self.running_flag，外部可通过方法修改终止循环
• 串口只初始化一次，避免重复打开关闭的资源浪费
• 线程设置为守护线程，即使主线程退出也会自动终止
• 增加串口异常捕获和最终关闭逻辑，避免资源泄漏
• 实现close_application方法，先终止数据采集再关闭应用

解决方案二：使用Kivy Clock替代线程（更符合Kivy异步机制） #

Kivy的Clock.schedule_interval是专门用于周期性任务的API，无需手动管理线程，更安全且适配Kivy的UI循环：

修改后的Python代码：

import serial
from kivy.app import App
from kivy.lang import Builder
from kivy.uix.screenmanager import Screen, ScreenManager
from kivy.properties import StringProperty
from kivy.clock import mainthread, Clock

class NaviWindow(Screen):
    temp_text = StringProperty("")
    humid_text = StringProperty("")
    schedule_id = None  # 保存调度任务ID，用于取消

    def __init__(self, **kwargs):
        super(NaviWindow, self).__init__(**kwargs)
        self.ser = None
        # 初始化串口，启动定时任务（60秒执行一次）
        try:
            self.ser = serial.Serial('COM3', 115200)
            self.schedule_id = Clock.schedule_interval(self.tmhm, 60)
        except Exception as e:
            print(f"串口初始化错误: {e}")

    def tmhm(self, dt):
        try:
            data = str(self.ser.readline().strip().decode())
            self.update_values(data.split(","))
        except Exception as e:
            print(f"读取数据错误: {e}")

    @mainthread
    def update_values(self, data):
        if len(data) >=2:
            self.temp_text = data[0]
            self.humid_text = data[1]

    def stop_recording(self):
        if self.schedule_id:
            Clock.unschedule(self.schedule_id)  # 取消定时任务
        if self.ser and self.ser.is_open:
            self.ser.close()

class WindowManager(ScreenManager):
    pass

kys = Builder.load_file("temp4.kv")

class MyMainApp(App):
    def build(self):
        return kys

    def close_application(self):
        current_screen = self.root.current_screen
        if isinstance(current_screen, NaviWindow):
            current_screen.stop_recording()
        self.stop()

if __name__ == "__main__":
    MyMainApp().run()

KV代码和方案一一致。

方案优势：

• 无需手动管理线程，避免线程同步问题
• 完全适配Kivy的UI循环，不会出现线程冲突
• 取消定时任务更简单，直接调用Clock.unschedule即可

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