Python：在使用多处理时更新 Tkinter

Posted 2023-02-16

【中文标题】Python：在使用多处理时更新 Tkinter

【英文标题】：Python: Update Tkinter while using Multiprocessing

【发布时间】：2021-12-09 11:10:18

【问题描述】：

我遇到以下问题：

我正在尝试使用 concurrent.futures.ProcessPoolExecutor() 或类似的东西，并在 tkinter 小部件上显示每个进程的进度。

有这个答案： Python Tkinter multiprocessing progress 但我不能让它工作。

我的代码的以下简化版本似乎仅在使用我不想要的 ThreadPoolExecutor() 时才有效。

提前感谢您的帮助！

import concurrent.futures
import tkinter
import tkinter.ttk
import multiprocessing
import random
import time


class App:
    def __init__(self, root):
        self.root = root

        self.processes = 5
        self.percentage = []
        self.changing_labels = []
        self.queues = []
        self.values = []

        for i in range(self.processes):
            temp_percentage = tkinter.StringVar()
            temp_percentage.set("0 %")
            self.percentage.append(temp_percentage)

            temp_changing_label = tkinter.Label(self.root, textvariable=temp_percentage)
            temp_changing_label.pack()
            self.changing_labels.append(temp_changing_label)

            self.queues.append(multiprocessing.Queue())
            # Just same values that I want to do calculations on
            temp_value = []
            for ii in range(12):
                temp_value.append(random.randrange(10))
            self.values.append(temp_value.copy())

        self.start_processing()

    def start_processing(self):
        def save_values(my_values):     # Save my new calculated values on the same file or different file
            with open(f"example.txt", "a") as file:
                for v in my_values:
                    file.write(str(v))
                    file.write(" ")
                file.write("\n")

        def work(my_values, my_queue):  # Here I do all my work
            # Some values to calculate my progress so that I can update my Labels
            my_progress = 0
            step = 100 / len(my_values)
            # Do some work on the values
            updated_values = []
            for v in my_values:
                time.sleep(0.5)
                updated_values.append(v + 1)

                my_progress += step
                my_queue.put(my_progress)   # Add current progress to queue

            save_values(updated_values)     # Save it before exiting

        # This Part does no work with ProcessPoolExecutor, with ThreadPoolExecutor it works fine
        with concurrent.futures.ProcessPoolExecutor() as executor:
            results = [executor.submit(work, self.values[i], self.queues[i])
                       for i in range(self.processes)]
            # Run in a loop and update Labels or exit when done
            while True:
                results_done = [result.done() for result in results]

                if False in results_done:
                    for i in range(self.processes):
                        if results_done[i] is False:
                            if not self.queues[i].empty():
                                temp_queue = self.queues[i].get()
                                self.percentage[i].set(f"temp_queue:.2f %")
                        else:
                            self.percentage[i].set("100 %")
                        self.root.update()
                else:
                    break
        # Close window at the very end
        self.root.destroy()


def main():  # Please do not change my main unless it is essential
    root = tkinter.Tk()
    my_app = App(root)
    root.mainloop()


if __name__ == "__main__":
    main()

【问题讨论】：

“不起作用”不是对问题的有用描述。此外，说它与 ThreadPoolExecutor() 一起“似乎有效”表明您无法真正判断它是否有效。您需要准确解释会发生什么以及您预期会发生什么。

【参考方案1】：

问题是由新进程内引发的异常引起的。要查看此异常，您可以调用Future.result 函数，该函数会引发异常，例如

print([result.result() for result in results])

这给出了错误AttributeError: Can't pickle local object 'App.start_processing.<locals>.work'，所以问题是work 是在另一个函数中定义的。

将work 和save_values 函数移出start_processing 方法后，错误变为RuntimeError: Queue objects should only be shared between processes through inheritance。这可以通过使用multiprocessing.Manager 创建队列来解决：

class App:
    def __init__(self, root):
        ...

        with multiprocessing.Manager() as manager:
            for i in range(self.processes):
                ...

                self.queues.append(manager.Queue())
            ...
            self.start_processing()

现在可以删除print([result.result() for result in results]) 调试行。 完整代码如下：

import concurrent.futures
import tkinter
import tkinter.ttk
import multiprocessing
import random
import time


class App:
    def __init__(self, root):
        self.root = root

        self.processes = 5
        self.percentage = []
        self.changing_labels = []
        self.queues = []
        self.values = []

        with multiprocessing.Manager() as manager:
            for i in range(self.processes):
                temp_percentage = tkinter.StringVar()
                temp_percentage.set("0 %")
                self.percentage.append(temp_percentage)

                temp_changing_label = tkinter.Label(self.root, textvariable=temp_percentage)
                temp_changing_label.pack()
                self.changing_labels.append(temp_changing_label)

                self.queues.append(manager.Queue())
                # Just same values that I want to do calculations on
                temp_value = []
                for ii in range(12):
                    temp_value.append(random.randrange(10))
                self.values.append(temp_value.copy())

            self.start_processing()

    @staticmethod
    def save_values(my_values):  # Save my new calculated values on the same file or different file
        with open(f"example.txt", "a") as file:
            for v in my_values:
                file.write(str(v))
                file.write(" ")
            file.write("\n")

    @classmethod
    def work(cls, my_values, my_queue):  # Here I do all my work
        # Some values to calculate my progress so that I can update my Labels
        my_progress = 0
        step = 100 / len(my_values)
        # Do some work on the values
        updated_values = []
        for v in my_values:
            time.sleep(0.5)
            updated_values.append(v + 1)

            my_progress += step
            my_queue.put(my_progress)  # Add current progress to queue

        cls.save_values(updated_values)  # Save it before exiting

    def start_processing(self):

        # This Part does no work with ProcessPoolExecutor, with ThreadPoolExecutor it works fine
        with concurrent.futures.ProcessPoolExecutor() as executor:
            results = [executor.submit(self.work, self.values[i], self.queues[i])
                       for i in range(self.processes)]
            # Run in a loop and update Labels or exit when done
            while True:
                results_done = [result.done() for result in results]

                if False in results_done:
                    for i in range(self.processes):
                        if results_done[i] is False:
                            if not self.queues[i].empty():
                                temp_queue = self.queues[i].get()
                                self.percentage[i].set(f"temp_queue:.2f %")
                        else:
                            self.percentage[i].set("100 %")
                        self.root.update()
                else:
                    break
        # Close window at the very end
        self.root.destroy()


def main():  # Please do not change my main unless it is essential
    root = tkinter.Tk()
    my_app = App(root)
    root.mainloop()


if __name__ == "__main__":
    main()

PS：我会推荐not all(results_done) 而不是False in results_done，以及not results_done[i] 而不是results_done[i] is False。