在另一个进程正在进行时如何显示加载进度条？
原创在另一个进程正在进行时如何显示加载进度条？
原创
华科云商小徐
发布于 2025-02-11 10:27:06
发布于 2025-02-11 10:27:06
代码可运行
文章被收录于专栏：小徐学爬虫
运行总次数：0
代码可运行
在 Tkinter 中，如果你希望在另一个进程（例如在后台运行的任务）正在进行时显示加载进度条，可以使用 ttk.Progressbar 组件来显示进度条，并通过使用 多线程 来避免界面冻结。
下面我将展示如何在后台运行一个耗时任务时，同时显示进度条，直到任务完成。
1、问题背景
在wxPython中，正在构建一个RSS feed聚合器客户端，其中一个想要实现的功能是刷新函数，该函数可以刷新feed并显示任何新文章。但是，每次调用刷新函数时，都必须重新抓取所有网络文章并显示在屏幕上，这通常需要大约6-7秒。因此，创建了一个带有加载进度条的wx.Dialog类，持续7秒。希望在抓取所有文章时显示此对话框和进度条。通过threading模块尝试了这种方法，但无济于事。对话框弹出，但进度条在所有事情都发生后才开始。
2、解决方案
2.1 异步更新进度条
需要从执行抓取的线程向主应用程序发送消息，以告诉它更新进度条。这意味着将使用wxPython的线程安全方法之一：
- wx.CallAfter
- wx.CallLater
- wx.PostEvent
可以使用pubsub结合CallAfter来实现。可以使用pubsub发布一条消息给对话框，该对话框需要有一个监听器。
2.2 代码示例
以下是一个示例应用程序，演示如何从线程中定期更新进度条小组件（适用于wxPython 2.8）：
import time
import wx
from threading import Thread
from wx.lib.pubsub import Publisher
########################################################################
class TestThread(Thread):
"""Test Worker Thread Class."""
#----------------------------------------------------------------------
def __init__(self):
"""Init Worker Thread Class."""
Thread.__init__(self)
self.start() # start the thread
#----------------------------------------------------------------------
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
for i in range(20):
time.sleep(1)
wx.CallAfter(Publisher().sendMessage, "update", "")
########################################################################
class MyProgressDialog(wx.Dialog):
""""""
#----------------------------------------------------------------------
def __init__(self):
"""Constructor"""
wx.Dialog.__init__(self, None, title="Progress")
self.count = 0
self.progress = wx.Gauge(self, range=20)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.progress, 0, wx.EXPAND)
self.SetSizer(sizer)
# create a pubsub receiver
Publisher().subscribe(self.updateProgress, "update")
#----------------------------------------------------------------------
def updateProgress(self, msg):
""""""
self.count += 1
if self.count >= 20:
self.Destroy()
self.progress.SetValue(self.count)
########################################################################
class MyForm(wx.Frame):
#----------------------------------------------------------------------
def __init__(self):
wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial")
# Add a panel so it looks the correct on all platforms
panel = wx.Panel(self, wx.ID_ANY)
self.btn = btn = wx.Button(panel, label="Start Thread")
btn.Bind(wx.EVT_BUTTON, self.onButton)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5)
panel.SetSizer(sizer)
#----------------------------------------------------------------------
def onButton(self, event):
"""
Runs the thread
"""
btn = event.GetEventObject()
btn.Disable()
TestThread()
dlg = MyProgressDialog()
dlg.ShowModal()
btn.Enable()
#----------------------------------------------------------------------
# Run the program
if __name__ == "__main__":
app = wx.PySimpleApp()
frame = MyForm().Show()
app.MainLoop()对于wxPython 2.9，pubsub已更新为使用新的pubsub API。以下示例适用于wxPython 2.9：
import time
import wx
from threading import Thread
from wx.lib.pubsub import pub
########################################################################
class TestThread(Thread):
"""Test Worker Thread Class."""
#----------------------------------------------------------------------
def __init__(self):
"""Init Worker Thread Class."""
Thread.__init__(self)
self.start() # start the thread
#----------------------------------------------------------------------
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
for i in range(20):
time.sleep(1)
wx.CallAfter(pub.sendMessage, "update", msg="")
########################################################################
class MyProgressDialog(wx.Dialog):
""""""
#----------------------------------------------------------------------
def __init__(self):
"""Constructor"""
wx.Dialog.__init__(self, None, title="Progress")
self.count = 0
self.progress = wx.Gauge(self, range=20)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.progress, 0, wx.EXPAND)
self.SetSizer(sizer)
# create a pubsub receiver
pub.subscribe(self.updateProgress, "update")
#----------------------------------------------------------------------
def updateProgress(self, msg):
""""""
self.count += 1
if self.count >= 20:
self.Destroy()
self.progress.SetValue(self.count)
########################################################################
class MyForm(wx.Frame):
#----------------------------------------------------------------------
def __init__(self):
wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial")
# Add a panel so it looks the correct on all platforms
panel = wx.Panel(self, wx.ID_ANY)
self.btn = btn = wx.Button(panel, label="Start Thread")
btn.Bind(wx.EVT_BUTTON, self.onButton)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5)
panel.SetSizer(sizer)
#----------------------------------------------------------------------
def onButton(self, event):
"""
Runs the thread
"""
btn = event.GetEventObject()
btn.Disable()
TestThread()
dlg = MyProgressDialog()
dlg.ShowModal()
btn.Enable()
#----------------------------------------------------------------------
# Run the program
if __name__ == "__main__":
app = wx.App(False)
frame = MyForm().Show()
app.MainLoop()最终我们可以根据实际需要调整进度条的更新频率，或者更复杂的任务逻辑。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
腾讯云开发者
