带有tkinter端按按钮执行的Python多处理

Question

我想用tkinter实现多处理的生产者消费者模式。我想通过按下按钮开始执行过程，并通过按另一个按钮来结束进程的执行。

到目前为止我所做的是：

import multiprocessing as mp
import random
import tkinter as tk
import logging

class MainApp(tk.Frame):
    def __init__(self, parent, *args, **kwargs):
        super().__init__(parent, *args, **kwargs)
        self.parent = parent   
    
        self.startbtn = tk.Button(self, text="Start MP", command=self.onstart)
        self.stopbtn = tk.Button(self, text="Stop MP", command=self.onstop)
    
        self.columnconfigure(index=0,weight=1)
        self.startbtn.grid(row=0, column=0, sticky="ew", pady=(25,0))
        self.stopbtn.grid(row=1, column=0, sticky="ew", pady=(25,25))
        
    def onstart(self):
        print("Start button clicked!")
        
        self.tasks = mp.Queue()
        
        self.producer = Producer(self.tasks)
        self.consumer = Consumer(self.tasks)
        
        self.producer.start()
        self.consumer.start()
        
        self.startbtn.config(state="disabled")
        self.stopbtn.config(state="active")
        
    def onstop(self):
 
        self.producer.event.set()
        self.consumer.event.set()  
        
        self.producer.join()
        self.consumer.join()
           
        self.producer.close()
        self.consumer.close()
        
        self.startbtn.config(state="active")
        self.stopbtn.config(state="disabled")
        

class Consumer(mp.Process):
    def __init__(self, taskqueue):
        super().__init__()
        self.taskqueue = taskqueue
        self.event = mp.Event()
        
        
    def run(self):
        procname = self.name
        print('{}'.format("Consumer started!"))
        while True:
            nexttask = self.taskqueue.get()
            print('Task:{}'.format(nexttask))
            if self.event.is_set():         
                print('{}:Exiting'.format(procname))
                break
        return
            
            
class Producer(mp.Process):
    def __init__(self, taskqueue):
        super(Producer, self).__init__()
        self.taskqueue = taskqueue
        self.event = mp.Event()  

        
    def run(self):
        procname = self.name
        print('{}'.format("Producer started!"))
        while True:
            task = random.random()
            self.taskqueue.put(task)
            if self.event.is_set():
                print('{}:Exiting'.format(procname))
                break
        return

if __name__ == '__main__':
    mlogger = mp.log_to_stderr()
    #mlogger.setLevel(mp.SPDEBUG)
    logging.basicConfig(level=logging.DEBUG)
    root = tk.Tk()
    root.geometry("600x400")
    root.columnconfigure(index=0, weight=1)
    root.rowconfigure(index=0, weight=1)
    mapp = MainApp(root)
    mapp.grid(row=0, column=0, sticky="nsew")

    root.mainloop()

但是，我无法通过按下“onstop万亿”并调用onstop函数来完成进程的执行。onstop函数挂起消息(队列线程上有死锁)：

Consumer-2:Exiting
Producer-1:Exiting
[INFO/Consumer-2] process shutting down
[DEBUG/Consumer-2] running all "atexit" finalizers with priority >= 0
[INFO/Producer-1] process shutting down
[DEBUG/Producer-1] running all "atexit" finalizers with priority >= 0
[DEBUG/Consumer-2] running the remaining "atexit" finalizers
[INFO/Consumer-2] process exiting with exitcode 0
[DEBUG/Producer-1] telling queue thread to quit
[DEBUG/Producer-1] running the remaining "atexit" finalizers
[DEBUG/Producer-1] joining queue thread

是否可以通过按下tkinter中的按钮来结束这些过程(生产者、消费者和队列)？我尝试了我在互联网上发现的所有关于这个话题的东西，但都没有成功。

Answer 1

如果有一个Queue等待发送或接收一个项目，它可以阻止进程加入。尝试清除队列是一种很好的做法，这可以通过多种方式完成。我会设置它，使生产者获得退出的信号，并将该信号作为队列中的最后一项发送给消费者：

from multiprocessing import Process, Event, Queue
from queue import Empty, Full

PUT_TIMEOUT = 1
GET_TIMEOUT = 1

class STOPFLAG: pass  #unique value as a flag for exiting

def producer(queue, event):
    while not event.is_set():
         try:
             queue.put("item", True, PUT_TIMEOUT)
         except Full:
             pass #handle queue is full
    queue.put(STOPFLAG())
    
def consumer(queue):
    while True:
        try:
            item = queue.get(True, GET_TIMEOUT)
        except Empty:
            continue #maybe do some housekeeping if no item is available from the producer right now
        if isinstance(item, STOPFLAG):
            return

在技术上，您还可以强制队列在后台使用守护进程线程来放置和获取项目，这样在进程结束时任何剩余的工作都会被删除。但是，您必须绝对确定，只要把所有东西都放在中间就可以了，而且您也不介意队列另一端出现的可能是部分或无效的对象：

def producer(queue, event):
    queue.cancel_join_thread()
    while not event.is_set():
         try:
             queue.put("item", True, PUT_TIMEOUT)
         except Full:
             pass #handle queue is full
    
def consumer(queue, event):
    queue.cancel_join_thread()
    while  not event.is_set():
        try:
            item = queue.get(True, GET_TIMEOUT)
        except Empty:
            continue #maybe do some housekeeping if no item is available from the producer