我有以下问题:
我正在尝试使用concurrent.futures.ProcessPoolExecutor()或任何类似的方法，并在tkinter小部件上显示每个进程的进度。
有这样的答案:Python Tkinter多进程，但我不能完全使它工作。
下面的简化版本的代码似乎只在使用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()
问题是由新进程内部抛出的异常引起的。要查看此异常，您可以调用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。