问题描述
我创建了一个 tkinter gui，它在单击某些按钮时会执行一些耗时的操作.在我的案例中，有两个主要问题会导致 tkinter 以不一致的方式崩溃.
i have created a tkinter gui that does some time consuming operations when clicking on some buttons. There are 2 main problems in my case that i think cause tkinter to crash in an unconssistent way.
1) 我想将新线程的一些输出记录到在主线程上运行的滚动文本小部件中，我认为这不是一个好主意.现在我通过将 TextHandler 对象发送到线程来做到这一点(请参阅我的代码)
1) I want to log some output from the new threads into a scrolled text widget that runs on the main thread, which i Think is not a good idea. Right now i do that by having a TextHandler object send to the threads (see my code)
2) 为了避免从新线程登录到主线程，我想知道线程何时完成，然后从主线程登录到文本小部件但主循环如何知道，线程何时完成是在按钮按下时调用的函数中创建的吗?
2) In order to avoid logging from the new thread to the main thread, i want to know when the thread is done and then log from the main thread into the text widget but how does the main loop knows, when the thread is created in the function called from the Button press?
class TextHandler(logging.Handler):
"""This class allows you to log to a Tkinter Text or ScrolledText widget"""
def __init__(self, text):
# run the regular Handler __init__
logging.Handler.__init__(self)
# Store a reference to the Text it will log to
self.text = text
def emit(self, record):
msg = self.format(record)
def append():
self.text.configure(state='normal')
self.text.insert(tkinter.END, msg + '\n')
self.text.configure(state=DISABLED)
# Autoscroll to the bottom
self.text.yview(tkinter.END)
# This is necessary because we can't modify the Text from other threads
self.text.after(0, append)
class GUI(Frame):
def __init__(self, parent, logger, st):
....initialize stuff...
def _startComputations(self):
if (self._paths == ""):
self._logger.warn("No drive paths found, please add a path")
else:
if (self._flexinput == ""):
self._logger.warn(self._flexinput)
self._logger.warn("No file path,continuing with standard KM estimation")
self.myThread = StandardUsecases(self._paths, self._input,
self._logger, self._standard_choices,
self._numberOfoccurences_usecases,
self._all_logs, self._unique_countries,
self.bt, self.bt2, self.bt3,
self._flexinput)
self.myThread.start()
self._all_jsons = self.myThread._original_json_list
self._general_json = self.myThread._general_json_data_list
def _KM_Button(self):
self._all_logs[:] = []
self.bt = Button(self.frame6, text='1-Main Usecases', font = "Helvetica 11 bold italic",
command = self._startComputations,relief = RAISED, bd= 6, bg = "pale green", fg = 'black')
self.bt.pack(side = LEFT)
def initGUI(self):
self.parent.title("DC Statistics calculator")
self.pack(fill=BOTH, expand=True)
self._Drive_Paths()
self._Flexray_Path()
#self._UserInput()
self._KM_Button()
self._countButton()
self._exportButton()
self._helpButton()
def main():
root = Tk()
root.geometry("700x800+400+400")
st = scrolledtext.ScrolledText(root, state='disabled')
st.configure(font="Times 13")
st.pack(side=BOTTOM, fill='both',expand='yes')
text_handler = TextHandler(st)
logger = logging.getLogger()
logger.addHandler(text_handler)
app = GUI(root, logger, st)
root.mainloop()
if __name__ == '__main__':
main()
所以按下按钮后，_startComputations 函数将被调用，并在那里创建线程.我发送 _logger 对象，以便我可以在新线程上登录到 ScrolledText 小部件.但大多数情况下，我会遇到类似python.exe 停止工作"或使用共享对象调用 tcl_appendlimitedtoobj 之类的崩溃.
So after pressing the button, the _startComputations function will be called, and there the thread will be created. I send the _logger object so that i can log to the ScrolledText widget while i am on the new thread. But most of the times i get crashes like "python.exe stopped working" or tcl_appendlimitedtoobj called with shared object.
如果我不想从新线程登录，主循环如何知道新线程是否完成，因为新线程是在按下按钮后调用的函数创建的?
In the case i do not want to log from the new thread, how can the main loop know if the new thread is done, since the new thread is created withid the function called after the button press?
谢谢
推荐答案
正如我在评论中所说的 - 您的错误发生是因为您将 text 小部件与 TextHandler 到一个单独的线程.TextHandler 的一些方法会转换你的 text 小部件的方法(例如 emit)，这些方法需要从主线程中进行转换.
As I said in comments - your error occures because you passing your text widget alongside with instance of TextHandler to a separate thread. And some methods of TextHandler casts methods of your text widget (e.g. emit), whom need to be casted from main thread.
解决您的问题最简单的解决方案是 queue 对象.您不能从其他线程转换 tkinter 方法(例如，向小部件写入内容)，但您可以填充 队列！这里的主要思想 - 在任务的任何线程存在时不断检查 queue.为了不断检查队列，我们需要一个after循环"和list任务:
The simpliest solution to overcome your problem is a queue object. You can't cast tkinter methods (e.g. write something to a widget) from other threads, but you can populate a queue! The main idea here - continiously check a queue while anyone of a task's threads exists. To continiously check the queue we need an after "loop" and list of tasks:
def listen(self, force_start=False):
# "after" loop - listener
self.listen_queue()
if self.task_list or force_start:
print('Listener: Listen')
self.after(100, self.listen)
else:
print('Listener: Off')
我们试图在其中从队列中提取一些东西代码>:
Inside whom we trying to pull something from a queue:
def listen_queue(self):
# listen queue
while self.log_queue.qsize():
try:
self.logger.warning(self.log_queue.get())
except queue.Empty:
pass
当然，您可以从应用程序一开始就开始监听 queue，通常，您的循环如下所示:
Of course you can start to listen a queue right from the start of your application and, in general, your loop would look like this:
def listen(self, force_start=False):
# "after" loop - listener
self.listen_queue()
self.after(100, self.listen)
但同样，如果您想对线程进行更多控制 - list 使用它们是个好主意！我想现在你已经掌握了模式，所以这里有一个完整的例子:
But again, if you want more control over threads - list with them is a good idea! I think that now you got the pattern, so here's a complete example:
# imports
try:
import tkinter as tk # Python 3
import queue
except ImportError:
import Tkinter as tk # Python 2
import Queue as queue
import logging
import threading
import random
import string
# classes
class TextHandler(logging.Handler):
def __init__(self, text):
# run the regular Handler __init__
logging.Handler.__init__(self)
# Store a reference to the Text it will log to
self.text = text
def emit(self, record):
msg = self.format(record)
self.text.configure(state='normal')
self.text.insert(tk.END, msg + '\n')
self.text.configure(state=tk.DISABLED)
self.text.yview(tk.END)
class App(tk.Tk):
# common tk app
def __init__(self):
# initiate root
tk.Tk.__init__(self)
self.resizable(width=False, height=False)
# initiate widgets
self.spawn_task_button = tk.Button(self, text='Spawn Task', command=self.spawn_task)
self.spawn_task_button.pack(expand=True, fill='x')
self.quit_button = tk.Button(self, text='Quit', command=self.close_app)
self.quit_button.pack(expand=True, fill='x')
self.text_console = tk.Text(self, bg='black', fg='white')
self.text_console.pack(expand=True, fill='both')
# initiate queue, task list, logger
self.logger = logging.getLogger()
self.logger.addHandler(TextHandler(self.text_console))
self.log_queue = queue.Queue()
self.task_list = []
# initiate events and protocols
self.protocol('WM_DELETE_WINDOW', self.close_app)
def put_line_to_queue(self, log_line=''):
# put log line to queue
self.log_queue.put(log_line)
def listen_queue(self):
# listen queue
while self.log_queue.qsize():
try:
self.logger.warning(self.log_queue.get())
except queue.Empty:
pass
def listen(self, force_start=False):
# "after" loop - listener
self.listen_queue()
if self.task_list or force_start:
print('Listener: Listen')
self.after(100, self.listen)
else:
print('Listener: Off')
def common_task(self, task_thread):
# example task wait + print
# add task to task_list
self.task_list.append(task_thread)
iteration_count = random.randint(1, 10)
task_numb = task_thread.name[-1:]
self.put_line_to_queue('\n*** Task %s: Spawned \t Iteration count: %d ***\n' % (task_numb, iteration_count))
for _ in range(iteration_count):
threading.Event().wait(1)
self.put_line_to_queue('Task %s: In Progress \t Iteration: %d \t Generated: %s' % (task_numb, _ + 1,
generate_smth()))
self.put_line_to_queue('Task %s: Completed\n' % task_numb)
# remove task from task_list
self.task_list.remove(task_thread)
def spawn_task(self):
# spawn another task
task = threading.Thread(target=lambda: self.common_task(task))
# "kick start" listener if task list is empty
if not self.task_list:
self.listen(force_start=True)
task.start()
def close_app(self):
# handle closing
if self.task_list:
# code to handle threads
# there're some threads in a list
self.put_line_to_queue('\n**** Cant quit right now! ****\n')
else:
self.destroy()
# functions
def generate_smth(size=6, chars=string.ascii_uppercase + string.digits):
# generate random
return ''.join(random.choice(chars) for _ in range(size))
# entry point
app = App()
app.mainloop()
这篇关于Tkinter python 在尝试登录主线程的新线程上崩溃的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！