如何在KivyMD应用的第二个线程中定时检测网络连接

问题背景 #

我开发了一款基于Kivy/KivyMD的双窗口应用，主窗口通过输入框监听用户输入并处理数据库操作，需要在后台定时检测设备联网状态，但多次尝试添加网络检测逻辑均未成功。

可行实现方案 #

以下是整合了网络定时检测功能的完整代码，核心思路是启动后台线程循环执行网络检测，通过Kivy的@mainthread装饰器确保UI更新操作在主线程执行：

完整代码 #

import fdb
import configparser
import threading
import time
import socket

from kivymd.app import MDApp
from kivy.properties import ListProperty
from kivy.lang import Builder
from kivy.clock import Clock
from kivy.uix.screenmanager import Screen, ScreenManager
from kivy.clock import mainthread
from kivymd.uix.pickers import MDDatePicker

config = configparser.ConfigParser()
config.read(r'C:\\Users\\Przemek\\Desktop\\RCP\\config.txt')   

RED_BACKGROUND = [.75, 0, 0, 1.0]
BLACK_BACKGROUND = [0.3, 0.3, 0.3, 1.0]
GREEN_BACKGROUND = [0.2, 0.3, 0.1, 1]
BLUE_BACKGROUND = [0, 0, 1, 1]

def check_connection():
    try:
        host = socket.gethostbyname('www.google.com')
        s = socket.create_connection((host, 80), 2)
        s.close()  # 记得关闭连接
        return True
    except:
        return False

KV = """
WindowManager:
    MainWindow:
    SecondWindow:

<MainWindow>:
    name: "MainWindow"
    canvas:
        Color:
            rgba: self.background
        Rectangle:
            pos: self.pos
            size: self.size

    BoxLayout:
        id: layout
        orientation: "vertical"
        pos: self.pos
        size: self.size
        
        Label:
            text: ''
            font_size: 1
        
        Image:
            source: 'logo.jpg'
            allow_stretch: True
            size_hint_y: None
            width: 80   

        Label:
            id: info
            text_size: self.size
            text: 'scan here'
            font_size: 26
            size_hint_y: None
            text_size: self.width, None
            height: self.texture_size[1]
            halign: 'center'

        Label:
            id: clock
            text: ''
            font_size: 60

        TextInput:
            id: kod
            text: ''
            multiline: False
            size_hint: (0, .5)
            font_size: 15
            
            #opacity: 0
            focus: True
            password: True
            text_validate_unfocus: False
            on_text_validate: root.update_label(kod.text)

        MDRaisedButton:
            text: 'button1'
            icon_color: 1, 1, 1, 1
            size_hint: 0.3, 0.5
            pos_hint: {"x":0.35, "top":1}
            on_release: 
                app.root.current = "SecondWindow"
            
        Label:
            id: wifi
            text: ''
            font_size: 16
            color: 1,1,1,1
        
        Label:
            text: 'gui apk'

<SecondWindow>
    name: "SecondWindow"

    MDBoxLayout:
        id: layout
        orientation: "vertical"
        pos: self.pos
        size: self.size
        md_bg_color: 0.3, 0.3, 0.3, 1.0
        
        Label:
            text: 'log in'
            font_size: 22

        MDRaisedButton:
            text: 'button1'
            size_hint: 0.3, 0.5
            pos_hint: {"x":0.35, "top":1}
            on_release: 
                app.show_date_picker()
                
        MDRaisedButton:
            text: 'exit'
            size_hint: 0.3, 0.5
            pos_hint: {"x":0.35, "top":1}

            on_release:
                root.manager.current = "MainWindow"
                root.manager.get_screen("MainWindow").ids["kod"].focus = True

        Label:
            text: 'gui apk'
        
"""

class MainWindow(Screen):
    background = ListProperty()
    
    def __init__(self, **kwargs):
        super(MainWindow, self).__init__(**kwargs)
        self.background = BLACK_BACKGROUND
        Clock.schedule_interval(self.UpdateClock, 1)
        # 启动后台网络检测线程，设置daemon=True确保主程序退出时线程自动终止
        threading.Thread(target=self.network_monitor, daemon=True).start()
        
    @mainthread
    def update_label(self, val):
        con = fdb.connect(
            port = int(config.get('global', 'port')),
            database = str(config.get('global', 'database')),
            host = str(config.get('global', 'host')),
            user = str(config.get('global', 'user')), 
            password = str(config.get('global', 'password')),
            charset = str(config.get('global', 'charset'))
        )
        cur = con.cursor()
             
        cur.execute("select KOD, FK_KRT_PRC, ID FROM TABLE")
        if len(val) > 0:
            print('main loop here')
        con.commit()
        con.close()
        self.ids.kod.text = ''
                
    def SetStartScreen(self, dt ):
        self.ids.info.text = 'scan'
        self.background = BLACK_BACKGROUND
        
    def UpdateClock(self, dt ):
        self.ids.clock.text = time.strftime("%H:%M:%S")

    def network_monitor(self):
        """后台线程循环检测网络状态"""
        while True:
            is_connected = check_connection()
            self.update_network_ui(is_connected)
            time.sleep(5)  # 每5秒检测一次，可根据需求调整
            
    @mainthread
    def update_network_ui(self, is_connected):
        """更新UI显示网络状态，必须在主线程执行"""
        if is_connected:
            self.ids.wifi.text = "✅ 已联网"
            self.ids.wifi.color = (0, 1, 0, 1)
            # 可选：联网时改变背景色
            # self.background = GREEN_BACKGROUND
        else:
            self.ids.wifi.text = "❌ 未联网"
            self.ids.wifi.color = (1, 0, 0, 1)
            # 可选：断网时改变背景色
            # self.background = RED_BACKGROUND
        
class SecondWindow(Screen):
    pass

class WindowManager(ScreenManager):
    pass

class MyMainApp(MDApp):
    def build(self):
        self.title = "gui apk"
        return Builder.load_string(KV)

    def on_save(self, instance, value, date_range):
        print(instance, value, date_range)
        pass

    def on_cancel(self, instance, value):
        pass

    def show_date_picker(self):
        date_dialog = MDDatePicker(mode="range")
        date_dialog.bind(on_save=self.on_save, on_cancel=self.on_cancel)
        date_dialog.open()

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

关键实现要点 #

• 后台线程启动：在MainWindow初始化时启动网络监测线程，设置daemon=True保证主程序退出时线程自动销毁
• 网络检测逻辑：check_connection函数通过连接Google服务器判断联网状态，检测完成后关闭socket连接避免资源泄漏
• UI更新限制：所有修改UI的操作必须用@mainthread装饰，确保在Kivy主线程执行，避免线程安全问题
• 定时检测间隔：通过time.sleep(5)设置每5秒检测一次，可根据实际需求调整间隔时长

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