我正在使用 wxPython 撰寫 Python 3.9 GUI 應用程式。該應用程式允許用戶輸入資訊并按下“計算”按鈕。這將啟動一個長時間運行的執行緒，該執行緒使用狀態更新更新主應用程式視窗。應用程式成功啟動計算執行緒并保持回應。
現在我想添加一個“中止”按鈕，當用戶選擇按下“中止”按鈕時，該按鈕將停止/取消/中止先前啟動的執行緒。
找了一段時間，在wxPython網站上找到了下面的代碼，里面有我要找的通用框架。它以更簡單的形式遵循我的應用程式的主要邏輯。它缺少正在進行的 GUI 狀態更新，所以我做了一個添加來模擬 GUI 狀態更新 - 請參閱下面代碼中的“<<<<<<”參考。
wx.PostEvent(self._notify_window, ResultEvent(i))
但是，在添加此行之后，當用戶在處理執行緒啟動時按下“停止”按鈕時，它似乎會中斷事件處理。在 3-4 秒后按下“停止”按鈕時，下面示例代碼中的計數執行緒繼續運行并忽略“停止”按鈕事件。
洗掉此行后，下面示例代碼中的“停止”功能將起作用，并且回圈中斷。
如何在計數執行緒運行時保持使用狀態更新更新 GUI 的功能，并讓“停止”按鈕終止正在運行的執行緒？
我確信有一些方法可以優化下面的代碼。請不要猶豫，分享您的想法/知識。謝謝你。
import time
from threading import *
import wx
# Button definitions
ID_START = wx.NewId()
ID_STOP = wx.NewId()
# Define notification event for thread completion
EVT_RESULT_ID = wx.NewId()
def EVT_RESULT(win, func):
"""Define Result Event."""
win.Connect(-1, -1, EVT_RESULT_ID, func)
class ResultEvent(wx.PyEvent):
"""Simple event to carry arbitrary result data."""
def __init__(self, data):
"""Init Result Event."""
wx.PyEvent.__init__(self)
self.SetEventType(EVT_RESULT_ID)
self.data = data
# Thread class that executes processing
class WorkerThread(Thread):
"""Worker Thread Class."""
def __init__(self, notify_window):
"""Init Worker Thread Class."""
Thread.__init__(self)
self._notify_window = notify_window
self._want_abort = 0
# This starts the thread running on creation, but you could
# also make the GUI thread responsible for calling this
#self.start()
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread. Simulation of
# a long process (well, 10s here) as a simple loop - you will
# need to structure your processing so that you periodically
# peek at the abort variable
for i in range(10):
if self._want_abort:
# Use a result of None to acknowledge the abort (of
# course you can use whatever you'd like or even
# a separate event type)
wx.PostEvent(self._notify_window, ResultEvent(None))
return
time.sleep(2)
wx.PostEvent(self._notify_window, ResultEvent(i)) <<<<<<< Added to simulate "status update" feedback of my application.
# Here's where the result would be returned (this is an
# example fixed result of the number 10, but it could be
# any Python object)
wx.PostEvent(self._notify_window, ResultEvent(10))
def abort(self):
"""abort worker thread."""
# Method for use by main thread to signal an abort
self._want_abort = 1
# GUI Frame class that spins off the worker thread
class MainFrame(wx.Frame):
"""Class MainFrame."""
def __init__(self, parent, id):
"""Create the MainFrame."""
wx.Frame.__init__(self, parent, id, 'Thread Test')
# Dumb sample frame with two buttons
wx.Button(self, ID_START, 'Start', pos=(0,0))
wx.Button(self, ID_STOP, 'Stop', pos=(0,50))
self.status = wx.StaticText(self, -1, '', pos=(0,100))
self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START)
self.Bind(wx.EVT_BUTTON, self.OnStop, id=ID_STOP)
# Set up event handler for any worker thread results
EVT_RESULT(self,self.OnResult)
# And indicate we don't have a worker thread yet
self.worker = None
def OnStart(self, event):
"""Start Computation."""
# Trigger the worker thread unless it's already busy
if not self.worker:
self.status.SetLabel('Starting computation')
self.worker = WorkerThread(self)
self.worker.start()
def OnStop(self, event):
"""Stop Computation."""
# Flag the worker thread to stop if running
if self.worker:
self.status.SetLabel('Trying to abort computation')
self.worker.abort()
def OnResult(self, event):
"""Show Result status."""
if event.data is None:
# Thread aborted (using our convention of None return)
self.status.SetLabel('Computation aborted')
else:
# Process results here
self.status.SetLabel('Computation Result: %s' % event.data)
# In either event, the worker is done
self.worker = None
class MainApp(wx.App):
"""Class Main App."""
def OnInit(self):
"""Init Main App."""
self.frame = MainFrame(None, -1)
self.frame.Show(True)
self.SetTopWindow(self.frame)
return True
if __name__ == '__main__':
app = MainApp(0)
app.MainLoop()
uj5u.com熱心網友回復：
你只是self.worker = None在錯誤的點上設定。
它應該只設定為Noneon Stop 或當它自然終止時，除此之外代碼沒問題。
例如
import time
from threading import *
import wx
# Button definitions
ID_START = wx.NewId()
ID_STOP = wx.NewId()
# Define notification event for thread completion
EVT_RESULT_ID = wx.NewId()
def EVT_RESULT(win, func):
"""Define Result Event."""
win.Connect(-1, -1, EVT_RESULT_ID, func)
class ResultEvent(wx.PyEvent):
"""Simple event to carry arbitrary result data."""
def __init__(self, data):
"""Init Result Event."""
wx.PyEvent.__init__(self)
self.SetEventType(EVT_RESULT_ID)
self.data = data
# Thread class that executes processing
class WorkerThread(Thread):
"""Worker Thread Class."""
def __init__(self, notify_window):
"""Init Worker Thread Class."""
Thread.__init__(self)
self._notify_window = notify_window
self._want_abort = 0
# This starts the thread running on creation, but you could
# also make the GUI thread responsible for calling this
#self.start()
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread. Simulation of
# a long process (well, 10s here) as a simple loop - you will
# need to structure your processing so that you periodically
# peek at the abort variable
for i in range(10):
if self._want_abort:
# Use a result of None to acknowledge the abort (of
# course you can use whatever you'd like or even
# a separate event type)
wx.PostEvent(self._notify_window, ResultEvent(None))
return
time.sleep(1)
wx.PostEvent(self._notify_window, ResultEvent(i)) # <<<<<<< Added to simulate "status update" feedback of my application.
# Here's where the result would be returned (this is an
# example fixed result of the number 10, but it could be
# any Python object)
time.sleep(2)
wx.PostEvent(self._notify_window, ResultEvent(10))
def abort(self):
"""abort worker thread."""
# Method for use by main thread to signal an abort
self._want_abort = 1
# GUI Frame class that spins off the worker thread
class MainFrame(wx.Frame):
"""Class MainFrame."""
def __init__(self, parent, id):
"""Create the MainFrame."""
wx.Frame.__init__(self, parent, id, 'Thread Test')
# Dumb sample frame with two buttons
wx.Button(self, ID_START, 'Start', pos=(0,0))
wx.Button(self, ID_STOP, 'Stop', pos=(0,50))
self.status = wx.StaticText(self, -1, '', pos=(0,100))
self.Bind(wx.EVT_BUTTON, self.OnStart, id=ID_START)
self.Bind(wx.EVT_BUTTON, self.OnStop, id=ID_STOP)
# Set up event handler for any worker thread results
EVT_RESULT(self,self.OnResult)
# And indicate we don't have a worker thread yet
self.worker = None
def OnStart(self, event):
"""Start Computation."""
# Trigger the worker thread unless it's already busy
if self.worker:
return
else:
self.status.SetLabel('Starting computation')
self.worker = WorkerThread(self)
self.worker.start()
self.status.SetLabel('Computation started')
def OnStop(self, event):
"""Stop Computation."""
# Flag the worker thread to stop if running
if self.worker:
self.status.SetLabel('Trying to abort computation')
self.worker.abort()
self.worker = None
else:
self.status.SetLabel('Computation not running')
def OnResult(self, event):
"""Show Result status."""
if event.data is None:
# Thread aborted (using our convention of None return)
self.status.SetLabel('Computation aborted')
else:
# Process results here
self.status.SetLabel('Computation Result: %s' % event.data)
if event.data == 10:
self.worker = None
self.status.SetLabel('Computation Finished')
class MainApp(wx.App):
"""Class Main App."""
def OnInit(self):
"""Init Main App."""
self.frame = MainFrame(None, -1)
self.frame.Show(True)
self.SetTopWindow(self.frame)
return True
if __name__ == '__main__':
app = MainApp(0)
app.MainLoop()
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/361153.html
上一篇：僅計算當前活動的執行緒