【发布时间】：2021-01-31 06:44:05
【问题描述】：
我正在做一个更大的项目，我有 2 个线程（相同的进程）和一个单独的进程。其中一个线程是 gui，另一个线程是哨兵线程，观察子进程，子进程正在用神经网络做一些繁重的工作。架构看起来有点像这样：
我需要能够取消神经网络的进程并分别结束哨兵线程。我创建了一个小例子，它展示了总体架构和我的方法。
from multiprocessing import Process, Queue
from threading import Thread
from time import sleep
class Worker(Process):
# The worker resembles the neural network. It does some calculations and shares
# the information via the queue.
def __init__(self, queue: Queue):
Process.__init__(self)
self.queue = queue
def run(self):
i = 0
while True:
self.queue.put(i)
i += 1
def stop(self):
# I used the stop function for trying out some things, like using a joinable
# queue and block execution as long as the queue is not empty, which is not
# working
self.queue.put(None)
self.terminate()
class Listener(Thread):
# This class resembles the sentinel thread. It checks in an infinite loop for
# messages. In the real application I send signals via the signals and slots
# design pattern to the gui and display the sent information.
def __init__(self):
Thread.__init__(self)
self.queue = Queue()
self.worker = Worker(self.queue)
def run(self):
self.worker.start()
while True:
data = self.queue.get()
if data is not None:
print(data)
else:
break
print("broken")
def stop(self):
self.worker.stop()
class System:
# This class resembles the gui
def __init__(self):
self.listener = Listener()
def start(self):
self.listener.start()
def stop(self):
self.listener.stop()
if __name__ == "__main__":
system = System()
system.start()
sleep(0.1)
system.stop()
有什么问题？
只要一个进程读取或写入队列，和/或队列没有被正确清空，一个或两个进程就会变成僵尸进程，这在某种意义上基本上是死锁。因此，我需要找到一种方法在终止进程时正确处理队列，从而使进程终止时不会出错。
到目前为止我所做的尝试：
-
对每个 task_done() 使用 Joinable Queue 和 join()
-
重写 SIGTERM 信号处理程序以等待队列被清空
-
使用可加入队列并且仅在 SIGTERM 信号处理程序中加入（）
结果：
-
处理速度大大下降，但终止工作正常
-
和 3. 终止不像我实现的那样工作 有时它有效，有时它没有。所以这种方法没有可靠的输出和知识
对 (3) 的尝试如下：
class Worker(Process):
def __init__(self, queue: Queue):
Process.__init__(self)
self.queue = queue
self.abort = False
self.lock = Lock()
signal(SIGTERM, self.stop)
def run(self):
i = 0
while True:
self.lock.acquire()
if self.abort:
break
else:
self.queue.put(i)
i += 1
self.lock.release()
exit(0)
def stop(self, sig, frame):
self.abort = True
self.queue.put(None)
self.queue.join()
exit(0)
【问题讨论】：
-
给系统加个心跳机制怎么样？让进程通信，它们每 N 秒就启动并运行一次。添加逻辑以在双方自 T 秒后未收到心跳时停止运行。
-
afaik 是队列中最大的问题。我需要工作进程停止将消息放入队列并让哨兵进程清理队列并获取所有消息。我还看不出心跳如何帮助解决这个问题。
-
为什么它不再有用了？ (1) 如果没有收到来自哨兵的心跳，worker 将停止将消息放入队列。 (2) 如果 Sentinel 没有收到来自 worker 的心跳，它会清理队列并获取所有消息。
-
如果工作类不使用主循环进行计算，而是进行长时间的顺序操作，您对实现它有何建议？
标签： python multithreading multiprocessing python-multiprocessing python-multithreading