如何在每次下载后更新Toga进度条？解决UI阻塞问题

问题核心是同步下载操作在UI线程执行，阻塞了UI事件循环，导致所有UI更新要等下载完全结束才生效。Toga提供了后台任务机制，可以把耗时的同步操作放到后台线程，同时安全地更新UI。

修改方案 #

• 使用add_background_task将下载逻辑移到后台线程，避免阻塞UI
• 所有UI元素的更新操作，必须通过UI线程的事件循环执行（用loop.call_soon_threadsafe），因为UI组件不是线程安全的

修改后的完整代码 #

"""
Downloader
"""
import pathlib, os
import toga
from toga.style import Pack
from toga.style.pack import COLUMN, ROW, LEFT, CENTER, TOP
from downloader.my_libs.downloader import downloader


WHITE = '#ffffff'
PRIMARY_COLOR = '#000000'
SECONDARY_COLOR = '#fbfbfb'
ACCENT_COLOR = '#80ff80'
BACKGROUND_COLOR = '#ffffff'


class Downloader(toga.App):
    def __init__(self, formal_name=None, app_id=None, app_name=None, id=None, icon=None, author=None, version=None, home_page=None, description=None, startup=None, windows=None, on_exit=None, factory=None):
        super().__init__(formal_name, app_id, app_name, id, icon, author, version, home_page, description, startup, windows, on_exit, factory)
        self.resource_folder = pathlib.Path(__file__).joinpath('../resources').resolve()
        self.url_filename = self.resource_folder.joinpath('url.txt')
        self.output_path = pathlib.Path(__file__).joinpath('../../../../files').resolve()
        # 修复原代码实例化错误：应使用导入的小写downloader类
        self.downloader = downloader(url='', output_path=self.output_path)
        self.number_of_items_downloaded = 0
        self.number_of_items_to_download = 0

    def startup(self):
        main_box = toga.Box(style=Pack(background_color=BACKGROUND_COLOR))
        box = toga.Box(style=Pack(background_color=BACKGROUND_COLOR))
        url_box = toga.Box()

        title_label = toga.Label(text='Downloader', style=Pack(text_align=CENTER, font_size=20, flex=1, background_color=BACKGROUND_COLOR, color=PRIMARY_COLOR))
        self.url_input = toga.TextInput(style=Pack(flex=4, font_size=12, background_color=SECONDARY_COLOR, color=PRIMARY_COLOR))
        self.download_button = toga.Button('Download', on_press=self.on_download, style=Pack(flex=1, font_size=12, padding_top=5, background_color=ACCENT_COLOR, color=PRIMARY_COLOR))
        self.info_box = toga.MultilineTextInput(readonly=True, style=Pack(flex=1, padding_top=10, font_size=12, background_color=SECONDARY_COLOR, color=PRIMARY_COLOR))
        self.progress_bar = toga.ProgressBar(max=100, value=0, style=Pack(padding_top=10, background_color=SECONDARY_COLOR, color=ACCENT_COLOR))

        box.add(title_label)
        url_box.add(self.url_input)
        box.add(url_box)
        box.add(self.download_button)
        box.add(self.info_box)
        box.add(self.progress_bar)
        main_box.add(box)

        box.style.update(direction=COLUMN, padding=50, flex=1)
        url_box.style.update(direction=ROW, flex=1, padding_top=5)

        self.main_window = toga.MainWindow(title=self.formal_name)
        self.main_window.content = main_box
        self.main_window.show()

    def on_download(self, widget):
        # 禁用按钮防止重复点击
        self.download_button.enabled = False
        self.progress_bar.value = 0
        self.number_of_items_downloaded = 0
        # 将下载任务加入后台线程
        self.add_background_task(self.download_files)
    
    def download_files(self):
        try:
            self.number_of_items_to_download = len(self.downloader.urls)

            for url in self.downloader.urls:
                try:
                    self.download_file(url)
                    self.update_progress()
                except Exception as exception:
                    error_msg = f'\nCould not download from: {url}\n{exception}\n\n'
                    # 在UI线程安全更新信息框
                    self.main_window.app.loop.call_soon_threadsafe(
                        lambda msg=error_msg: setattr(self.info_box, 'value', self.info_box.value + msg)
                    )
                    print('Could not download from:', url)
                    print(exception, '\n\n')
        finally:
            # 任务结束后重新启用按钮
            self.main_window.app.loop.call_soon_threadsafe(
                lambda: setattr(self.download_button, 'enabled', True)
            )

    def update_progress(self):
        progress = (self.number_of_items_downloaded / self.number_of_items_to_download) * 100
        # 在UI线程安全更新进度条
        self.main_window.app.loop.call_soon_threadsafe(
            lambda p=progress: setattr(self.progress_bar, 'value', p)
        )

    def download_file(self, url):
        title, filepath = self.downloader.download(url)
        download_msg = f'Downloaded {title} \nTo {filepath}\n\n'
        # 在UI线程安全更新信息框
        self.main_window.app.loop.call_soon_threadsafe(
            lambda msg=download_msg: setattr(self.info_box, 'value', self.info_box.value + msg)
        )
        self.number_of_items_downloaded += 1


def main():
    return Downloader()

关键修改说明 #

• 后台任务执行：on_download中用add_background_task替代直接调用，把下载逻辑移到后台线程，避免阻塞UI
• UI线程安全更新：所有修改info_box和progress_bar的操作，都用loop.call_soon_threadsafe包装，确保在UI线程执行
• 按钮状态控制：下载开始时禁用按钮，防止重复点击；任务结束后自动启用
• 实例化错误修复：原代码中错误地用自身类Downloader实例化下载器，改为导入的小写downloader类

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