Python+Kivy开发测速应用如何实现函数并行与UI实时更新？

Kivy的GUI渲染、事件响应全部在主线程运行，把耗时30秒的测速逻辑直接放在主线程执行时，主线程的事件循环会被阻塞，所有界面重绘、用户操作响应都会被暂停，自然会出现界面卡死无响应的情况，直到测速函数执行完成主线程才能恢复工作。

核心逻辑非常直接：

• 把耗时的测速逻辑全部放到独立子线程运行，避免阻塞主线程
• 子线程绝对不能直接修改Kivy的UI控件属性，所有界面更新（包括进度条变动、实时结果输出）都必须通过Kivy自带的Clock.schedule_once接口切回主线程执行，否则会出现随机崩溃、渲染异常问题。

两个需求的对应实现逻辑：

• 加载进度条：在测速流程的每个阶段（比如连接服务器、测延迟、测下载、测上传）计算当前进度值，通过Clock调度回主线程更新进度条的value属性；如果测速函数是黑盒拿不到精确进度，可以切换为Kivy进度条的不确定模式，显示往返滚动的加载动画即可
• 实时结果输出：子线程每拿到一段测速数据（瞬时速率、当前延迟、阶段结果），同样通过Clock调度回主线程，把文本追加到界面的Label控件上。

这里以常用的speedtest测速库举例，你可以直接替换成自己写的测速函数：

from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.progressbar import ProgressBar
from kivy.uix.label import Label
from kivy.uix.button import Button
from kivy.clock import Clock
import threading
import speedtest

class SpeedTestLayout(BoxLayout):
    def __init__(self, **kwargs):
        super().__init__(orientation="vertical", padding=20, spacing=15, **kwargs)
        # 实时结果展示文本
        self.status_label = Label(text="点击按钮开始测速", size_hint=(1, 0.6))
        self.add_widget(self.status_label)
        # 进度条，最大值设为100对应百分比进度
        self.progress_bar = ProgressBar(max=100, value=0, size_hint=(1, 0.1))
        self.add_widget(self.progress_bar)
        # 启动按钮
        self.start_btn = Button(text="开始测速", size_hint=(1, 0.2))
        self.start_btn.bind(on_press=self.start_speed_test)
        self.add_widget(self.start_btn)

    def update_ui(self, dt=None, progress=None, text=None):
        # 仅在主线程执行，专门处理所有UI更新
        if progress is not None:
            self.progress_bar.value = progress
        if text is not None:
            self.status_label.text += f"\n{text}"

    def run_speed_test(self):
        # 整个测速逻辑在子线程运行，禁止直接修改UI
        try:
            st = speedtest.Speedtest()
            # 阶段1：获取匹配的测速服务器
            Clock.schedule_once(lambda dt: self.update_ui(progress=10, text="正在拉取测速服务器列表..."))
            st.get_servers()
            best_server = st.get_best_server()
            Clock.schedule_once(lambda dt: self.update_ui(progress=25, text=f"已连接最优节点：{best_server['sponsor']} ({best_server['name']})"))
            
            # 阶段2：测试网络延迟
            Clock.schedule_once(lambda dt: self.update_ui(progress=40, text="正在测试网络延迟..."))
            ping_val = st.results.ping
            Clock.schedule_once(lambda dt: self.update_ui(progress=50, text=f"延迟测试完成：{ping_val:.2f} ms"))

            # 阶段3：测试下载速度，实时回传进度和瞬时速率
            def download_progress(current, total, *args):
                cur_progress = 50 + (current / total) * 25
                cur_speed = (current * 8 / 1024 / 1024) # 转成Mbps
                Clock.schedule_once(lambda dt: self.update_ui(progress=cur_progress, text=f"下载测试中：{cur_speed:.2f} Mbps"))
            Clock.schedule_once(lambda dt: self.update_ui(progress=50, text="开始测试下载速度..."))
            download_res = st.download(callback=download_progress) / 1024 / 1024 * 8
            Clock.schedule_once(lambda dt: self.update_ui(progress=75, text=f"下载速度测试完成：{download_res:.2f} Mbps"))

            # 阶段4：测试上传速度，实时回传进度和瞬时速率
            def upload_progress(current, total, *args):
                cur_progress = 75 + (current / total) * 25
                cur_speed = (current * 8 / 1024 / 1024) # 转成Mbps
                Clock.schedule_once(lambda dt: self.update_ui(progress=cur_progress, text=f"上传测试中：{cur_speed:.2f} Mbps"))
            Clock.schedule_once(lambda dt: self.update_ui(progress=75, text="开始测试上传速度..."))
            upload_res = st.upload(callback=upload_progress) / 1024 / 1024 * 8
            Clock.schedule_once(lambda dt: self.update_ui(progress=100, text=f"上传速度测试完成：{upload_res:.2f} Mbps\n测速全部结束"))

            # 测速完成恢复按钮可用
            Clock.schedule_once(lambda dt: setattr(self.start_btn, 'disabled', False))
        except Exception as e:
            Clock.schedule_once(lambda dt: self.update_ui(text=f"测速异常：{str(e)}"))
            Clock.schedule_once(lambda dt: setattr(self.start_btn, 'disabled', False))

    def start_speed_test(self, instance):
        # 点击按钮重置界面，启动子线程
        self.progress_bar.value = 0
        self.status_label.text = "测速初始化中..."
        self.start_btn.disabled = True
        # 设为守护线程，窗口关闭时线程自动退出
        test_thread = threading.Thread(target=self.run_speed_test, daemon=True)
        test_thread.start()

class SpeedTestApp(App):
    def build(self):
        return SpeedTestLayout()

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

替换自有测速函数时，只需要把run_speed_test里的speedtest相关逻辑替换成你自己的代码，在需要更新进度、输出结果的位置调用Clock.schedule_once触发update_ui即可。如果拿不到精确进度，可以把ProgressBar的max设为1，配合Clock.schedule_interval定时往返修改value，就能实现常见的不确定加载动画，测速结束后取消定时任务就行。

• 启动测速子线程必须加daemon=True，不然关闭应用窗口后，后台测速线程不会自动退出，会导致进程残留
• 别在子线程直接操作任何Kivy控件，哪怕只是改个文字颜色，都有概率触发崩溃，所有UI改动全走Clock.schedule_once回主线程执行
• 测速启动后要把开始按钮设为禁用，避免用户重复点击启动多个测速线程，测速完成或报错后再恢复按钮可用状态

界面效果参考 #

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