Kivy多线程应用异常：无数据交换时启动计算线程仍丢帧

问题背景 #

作为音乐人，正在开发一款用于模块化合成器的信号发生器（生成闸门、包络、控制信号），通过数模转换器输出电压控制合成器。选择Kivy作为GUI框架，看中其跨平台（支持平板、Raspberry Pi）及触摸屏适配能力，自身Python编程能力尚可。

由于应用需以8000样本/秒或更高的采样率生成信号，而Kivy默认帧率仅30fps，因此将信号计算逻辑放入独立线程，GUI仅负责参数调整。但启动计算线程后，即便线程间仅通过threading.Event控制启停、无任何数据交换，Kivy GUI仍出现明显帧丢失，拖动按钮卡顿严重，无法满足现场演出的流畅性需求。

复现代码 #

from kivy.app import App
from kivy.uix.floatlayout import FloatLayout
from kivy.uix.button import Button
from kivy.uix.behaviors import DragBehavior
import threading
from time import perf_counter

class ThreadingApp(App):

    stop_compute_thread = threading.Event()

    def __init__(self, **kwargs):
        super(ThreadingApp, self).__init__(**kwargs)
        self.sample_rate = 8000
        self.compute_thread_run = False

    def on_stop(self):
        # 关闭应用时停止子线程
        self.stop_compute_thread.set()

    def build(self):
        layout = FloatLayout()
        ThreadButton = Button(text="Start/Stop Compute Thread", size_hint=(.3, .15), pos_hint={"right": 1, "top":1})
        ThreadButton.bind(on_release=lambda x: self.thread_start_stop())
        layout.add_widget(ThreadButton)
        DragButton = DraggableButton(text="Dragable Button", size_hint=(None, None), size=(250, 150))
        layout.add_widget(DragButton)
        return layout

    def thread_start_stop(self):
        # 启动/停止计算线程
        if self.compute_thread_run:
            self.stop_compute_thread.set()
        else:
            threading.Thread(target=self.compute_thread).start()

    def compute_thread(self):
        print("Starting secondary compute thread.")
        self.compute_thread_run = True
        iteration = 0
        iteration_time = 1 / self.sample_rate
        thread_start = perf_counter()

        while True:
            iter_start = perf_counter()
            if self.stop_compute_thread.is_set():
                thread_stop = perf_counter()
                self.compute_thread_run = False
                self.stop_compute_thread.clear()
                print("Computed ", iteration, " iterations in ", thread_stop - thread_start, " seconds.")
                return
            self.compute_signal()
            while perf_counter() < iter_start + iteration_time:
                # 等待当前样本计算周期结束
                # 此处是卡顿根源？
                pass
            iteration += 1

    def compute_signal(self):
        # 信号计算占位方法，实际会包含信号生成及DA输出逻辑
        # 线程间共享变量将通过队列处理
        pass

class DraggableButton(DragBehavior, Button):
    def on_touch_move(self, touch):
        if self.collide_point(*touch.pos):
            self.center = touch.pos
            return True
        return super(DraggableButton, self).on_touch_move(touch)

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

问题根源 #

卡顿的核心原因是计算线程中的忙等待循环：

while perf_counter() < iter_start + iteration_time:
    pass

这个循环会让CPU核心持续处于100%负载状态，不断抢占CPU时间片，导致Kivy的GUI线程（负责渲染、触摸响应）无法获得足够的执行资源，最终出现帧丢失和操作卡顿。

解决方案 #

1. 替换忙等待为time.sleep()+短时间忙等待 #

直接使用time.sleep()可能存在精度问题（部分系统最小sleep精度约10ms），无法满足8000采样率（每125μs一次）的定时要求。可以先sleep掉大部分等待时间，剩余极短时间用忙等待补足，既减少CPU占用，又保证定时精度：

def compute_thread(self):
    print("Starting secondary compute thread.")
    self.compute_thread_run = True
    iteration = 0
    iteration_time = 1 / self.sample_rate
    thread_start = perf_counter()
    import time

    while True:
        iter_start = perf_counter()
        if self.stop_compute_thread.is_set():
            thread_stop = perf_counter()
            self.compute_thread_run = False
            self.stop_compute_thread.clear()
            print("Computed ", iteration, " iterations in ", thread_stop - thread_start, " seconds.")
            return
        self.compute_signal()
        # 计算剩余等待时间
        remaining = iter_start + iteration_time - perf_counter()
        if remaining > 0:
            # 先sleep掉大部分时间，留100μs做忙等待保证精度
            if remaining > 0.0001:
                time.sleep(remaining - 0.0001)
            # 最后短时间用忙等待
            while perf_counter() < iter_start + iteration_time:
                pass
        iteration += 1

2. 降低计算线程优先级（Unix/Linux） #

在Unix-like系统（如Raspberry Pi）中，可以降低计算线程的CPU优先级，让GUI线程优先获得资源：

import os

def compute_thread(self):
    # 降低线程优先级（数值越大优先级越低，范围-20到19）
    os.nice(10)
    print("Starting secondary compute thread.")
    # 后续逻辑不变...

3. 使用实时调度策略（Raspberry Pi/Linux，需root权限） #

如果需要更稳定的定时精度，可将计算线程设置为实时调度策略，同时保证GUI线程不受影响：

import os
import sched

def compute_thread(self):
    # 设置线程为FIFO实时调度，优先级1（需root权限）
    os.sched_setscheduler(0, sched.SCHED_FIFO, os.sched_param(1))
    print("Starting secondary compute thread.")
    # 后续逻辑不变...

总结 #

核心思路是避免CPU无意义的忙等待，通过合理的睡眠+精度补足、线程优先级调整等方式，让GUI线程获得足够的执行资源，同时保证信号生成的定时精度。针对Raspberry Pi这类嵌入式设备，实时调度策略能进一步提升稳定性。

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