当异步任务为 运行 时,如何防止 tkinter Gui 冻结?

How can I prevent a tkinter Gui from freezing while an async task is running?

我想用 tkinter 创建一个非阻塞的 Gui。到目前为止,我所看到的方式是,您可以像使用多进程一样进行操作。但是现在我有一个问题,我想用新创建的线程再次访问 gui 的主循环,我总是在这里出错。你能在两个线程之间来回跳转还是有其他不阻塞 Gui 的方法?

import asyncio
import tkinter as tk 
import multiprocessing as mp 

class pseudo_example():


    def app(self):
        self.root = tk.Tk()
        self.root.minsize(100,100)

        start_button = tk.Button(self.root, text="start", command=lambda: mp.Process(target=self.create_await_fun).start())
        start_button.pack()  #

        self.testfield = tk.Label(self.root, text="test")
        self.testfield.pack()

        #self.root.update_idletasks()
        self.root.mainloop()

    def create_await_fun(self):
        asyncio.run(self.await_fun())

    async def await_fun(self):
        self.root.update_idletasks()
        self.testfield["text"] = "start waiting"
        await asyncio.sleep(2)
        self.testfield["text"] = "end waiting"



if __name__ == '__main__':
    try:
        gui = pseudo_example()
        gui.app()
    except KeyboardInterrupt:
        print("Interrupted")
        sys.exit()

错误信息:

[xcb] 处理队列时序列号未知 [xcb] 很可能这是一个多线程客户端并且尚未调用 XInitThreads [xcb] 中止,对此感到抱歉。 XIO: X 服务器“:0”上的致命 IO 错误 0(成功) 在 401 个请求(已知已处理 401 个)后,剩余 0 个事件。 python3.8: ../../src/xcb_io.c:259: poll_for_event: 断言 `!xcb_xlib_threads_sequence_lost' 失败。

我知道存在 after() 方法,但我不知道如何在不启动 asyncio 任务的情况下将它与 asyncio 一起使用。 Asyncio 在最小示例中是不必要的,但我需要它用于另一个应用程序。

Tkinter 不支持 multi-task/multithreading。 您可以使用 mtTkinter,它在使用 multi-tasking 时提供线程安全:https://pypi.org/project/mttkinter/

您也可以使用队列系统在两个函数之间传输数据,但是您不能对跨越两个线程的 tkinter 对象执行任何操作。

不知道这是否有帮助。可能比使用 asyncio 更好。

ItzTheDodo.

我一般用队列,我给你放一个sample.In掩码你会 在状态栏上查看更改时间,该时间由单独的线程管理 并通过咨询线程尾部,查看 class 时钟,同时您可以使用不带掩码的各种按钮 freezing.I 认为您可以从这个示例中获得灵感。

#!/usr/bin/python3
import sys
import threading
import queue
import datetime
import time
import tkinter as tk
from tkinter import ttk
from tkinter import messagebox

class Clock(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

        self.queue = queue.Queue()
        self.check = True

    def stop(self):
        self.check = False

    def run(self):

        """Feeds the tail."""

        while self.check:
            s = "Astral date: "
            t = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            msg = "{0} {1}".format(s, t)
            time.sleep(1)
            self.queue.put(msg)
    
    def check_queue(self, obj):

        """Returns a formatted string representing time.
           obj in this case is the statusbar text"""

        while self.queue.qsize():
            try:
                x = self.queue.get(0)
                msg = "{0}".format(x)
                obj.set(msg)
            except queue.Empty:
                pass
            

class Main(ttk.Frame):
    def __init__(self, parent, ):
        super().__init__(name="main")

        self.parent = parent
        self.text = tk.StringVar()
        self.spins = tk.IntVar()
        self.option = tk.IntVar()
        self.check = tk.BooleanVar()
        self.values = ('Apple','Banana','Orange')
        self.status_bar_text = tk.StringVar()
        self.init_status_bar()
        self.init_ui()

    def init_status_bar(self):

        self.status = tk.Label(self,
                               textvariable=self.status_bar_text,
                               bd=1,
                               relief=tk.SUNKEN,
                               anchor=tk.W)
        self.status.pack(side=tk.BOTTOM, fill=tk.X)        
          
    def init_ui(self):

        f0 = ttk.Frame(self)
        f1 = ttk.Frame(f0,)

        ttk.Label(f1, text = "Combobox").pack()
        self.cbCombo = ttk.Combobox(f1,state='readonly',values=self.values)
        self.cbCombo.pack()
        
        ttk.Label(f1, text = "Entry").pack()
        self.txTest = ttk.Entry(f1, textvariable=self.text).pack()

        ttk.Label(f1, text = "Spinbox").pack()
        tk.Spinbox(f1, from_=0, to=15, textvariable= self.spins).pack()

        ttk.Label(f1, text="Checkbutton:").pack()
        ttk.Checkbutton(f1,
                       onvalue=1,
                       offvalue=0,
                       variable=self.check).pack()


        ttk.Label(f1, text="Radiobutton:").pack()
        for index, text in enumerate(self.values):
            ttk.Radiobutton(f1,
                            text=text,
                            variable=self.option,
                            value=index,).pack()
            

        ttk.Label(f1, text="Listbox:").pack()
        self.ListBox = tk.Listbox(f1)
        self.ListBox.pack()
        self.ListBox.bind("<<ListboxSelect>>", self.on_listbox_select)
        self.ListBox.bind("<Double-Button-1>", self.on_listbox_double_button)
        
              
        f2 = ttk.Frame(f0,)

        bts = [("Callback", 7, self.on_callback, "<Alt-k>"),
               ("Args", 0, self.on_args, "<Alt-a>"),
               ("kwargs", 1, self.on_kwargs, "<Alt-w>"),
               ("Set", 0, self.on_set, "<Alt-s>"),
               ("Reset", 0, self.on_reset, "<Alt-r>"),
               ("Close", 0, self.on_close, "<Alt-c>")]

        for btn in bts:
            ttk.Button(f2,
                       text=btn[0],
                       underline=btn[1],
                       command = btn[2]).pack(fill=tk.X, padx=5, pady=5)
            self.parent.bind(btn[3], btn[2])
            
        f1.pack(side=tk.LEFT, fill=tk.BOTH, expand=1)
        f2.pack(side=tk.RIGHT, fill=tk.Y, expand=0)
        f0.pack(fill=tk.BOTH, expand=1)

    def on_open(self):
        self.periodic_call()
        
    def on_callback(self, evt=None):

        print ("self.cbCombo = {}".format(self.cbCombo.get()))
        print ("self.text = {}".format(self.text.get()))
        print ("self.spins = {}".format(self.spins.get()))
        print ("self.check = {}".format(self.check.get()))
        print ("self.option = {}".format(self.option.get()))
        if self.ListBox.curselection():
            print("ListBox.curselection = {}".format(self.ListBox.curselection()[0]))
        else:
            print("{0}".format("No selected item on listbox"))

    def on_args(self, evt=None):

        print("args type: {}".format(type(self.master.args)))
        for p, i in enumerate(self.master.args):
            print(p, i)

    def on_kwargs(self, evt=None):

        print("kwargs type: {}".format(type(self.master.kwargs)))
        for k, v in self.master.kwargs.items():
            print("{0}:{1}".format(k,v))

    def on_reset(self, evt=None):
        self.text.set('')
        self.spins.set(0)
        self.check.set(0)

    def on_set(self, evt=None):
        self.cbCombo.current(1)
        self.text.set('qwerty')
        self.spins.set(42)
        self.check.set(1)
        self.option.set(1)
        self.ListBox.delete(0, tk.END)
        
        for i in self.values:
            s = "{0}".format(i,)
            self.ListBox.insert(tk.END, s)
     
        self.ListBox.selection_set(1)

    def on_listbox_select(self, evt=None):

        if self.ListBox.curselection():

            index = self.ListBox.curselection()
            s = self.ListBox.get(index[0])
            print("on_listbox_select: index = {0} values = {1}".format(index, s))

    def on_listbox_double_button(self, evt=None):

        if self.ListBox.curselection():
            index = self.ListBox.curselection()
            s = self.ListBox.get(index[0])
            print("on_listbox_double_button: index = {0} values = {1}".format(index, s))

    def periodic_call(self):
        """This funciont check the data returned from the clock class queue."""

        self.parent.clock.check_queue(self.status_bar_text)
        
        if self.parent.clock.is_alive():
            self.after(1, self.periodic_call)
        else:
            pass                    
        
    def on_close(self, evt=None):
        self.parent.on_exit()


class App(tk.Tk):
    """Main Application start here"""
    def __init__(self, *args, **kwargs):
        super().__init__()

        self.args = args
        self.kwargs = kwargs
        self.protocol("WM_DELETE_WINDOW", self.on_exit)
        self.set_style(kwargs["style"])  
        self.set_title(kwargs["title"])
        self.resizable(width=False, height=False)

        #start clock on a separate thread...
        self.set_clock()
        
        w = Main(self)
        w.on_open()
        w.pack(fill=tk.BOTH, expand=1)

    def set_clock(self,):
        self.clock = self.get_clock()
        self.clock.start()
        
    def get_clock(self,):
        """Instance the clock."""
        return Clock()
    
    def set_style(self, which):
        self.style = ttk.Style()
        self.style.theme_use(which)
        
    def set_title(self, title):
        s = "{0}".format(title)
        self.title(s)
        
    def on_exit(self):
        """Close all"""
        msg = "Do you want to quit?"
        if messagebox.askokcancel(self.title(), msg, parent=self):
            #stop the thread
            if self.clock is not None:
                self.clock.stop()
            self.destroy()

def main():

    args = []

    for i in sys.argv:
        args.append(i)

    #('winnative', 'clam', 'alt', 'default', 'classic', 'vista', 'xpnative')
    kwargs = {"style":"clam", "title":"Simple App",}

    app = App(*args, **kwargs)

    app.mainloop()

if __name__ == '__main__':
    main()            
    

关于 我解决了我的问题。我已经创建了一个额外的线程,以便 GUI 和函数都可以在其中执行它们。

class pseudo_example():


    def app(self,async_loop):
        self.root = tk.Tk()
        self.root.minsize(100,100)

        self.start_button = tk.Button(self.root, text="start", command=lambda: self.create_await_fun(async_loop))
        self.start_button.pack()

        self.testfield = tk.Label(self.root, text="output")
        self.testfield.pack()
        self.root.mainloop()

    def create_await_fun(self,async_loop):
        threading.Thread(target=self.asyncio_thread, args=(async_loop,)).start()
        self.start_button["relief"] = "sunken"
        self.start_button["state"] = "disabled"

    def asyncio_thread(self, async_loop):
        async_loop.run_until_complete(self.await_fun())

    async def await_fun(self):

        self.testfield["text"] = "start waiting"
        self.root.update_idletasks()

        await asyncio.sleep(2)

        self.testfield["text"] = "end waiting"
        self.root.update_idletasks()

        await asyncio.sleep(1)

        self.testfield["text"] = "output"
        self.root.update_idletasks()
        self.start_button["relief"] = "raised"
        self.start_button["state"] = "normal"


if __name__ == '__main__':
    gui = pseudo_example()
    async_loop = asyncio.get_event_loop()
    gui.app(async_loop)