python - threading.Thread 对抗 multiprocessing.Process

Question

所以最近我遇到了以下问题:我必须创建一个处理请求的服务器，以便在主进程使用这些值时更新某些值。 所以这里，服务器处理函数是在子进程中，我想停止的时候都无法停止。

为了测试在 threading.Thread 或 multiprocessing.Process 之间解决我的问题的最佳解决方案是什么，我制作了以下小程序:

import multiprocessing
import time
import threading

class SubProcess(multiprocessing.Process):

    def __init__(self, source):
        self.source = source
        super(SubProcess, self).__init__()

    def run(self):
        while 1:
            time.sleep(1) # Waiting for request...
            self.source.somevar["anotherkey"].append(5)
            print "My subprocess : ", id(self.source), id(self.source.somevar), self.source.somevar

class SubThread(threading.Thread):

    def __init__(self, source):
        self.source = source
        super(SubThread, self).__init__()

    def run(self):
        while 1:
            time.sleep(1) # Waiting for request...
            self.source.somevar["anotherkey"].append(5)
            print "My subthread : ", id(self.source), id(self.source.somevar), self.source.somevar

class Source:

    def __init__(self):
        self.somevar = {"akey": "THE key", "anotherkey": [5]}

    def start_process(self):
        self.process = SubProcess(self)
        self.process.start()

    def stop_process(self):
        self.process.terminate()

    def start_thread(self):
        self.thread = SubThread(self)
        self.thread.start()

    def stop_thread(self):
        # self.thread.stop() # What the hell should i put here
        pass

s = Source()

s.start_process()
time.sleep(2)
print "End : ", id(s), id(s.somevar), s.somevar
s.stop_process()

s.start_thread()
time.sleep(2)
print "End : ", id(s), id(s.somevar), s.somevar
s.stop_thread() # Obviously, thread never ends...

所以threading.Thread修改了原来的s.somevar但我无法阻止它，而multiprocessing.Process不修改原来的 s.somevar 但我可以阻止它。

我正在寻找一种解决方案，可以停止线程(使用 SIGTERM)，并且线程可以修改原始类Source，使用标准库 。有什么解决办法吗？

Answer 1

要杀死线程，您需要在子线程和主线程之间进行一些协作。在示例代码中，您可以使用 threading.Event:

class SubThread(threading.Thread):

    def __init__(self, source):
        self.source = source
        self.should_stop = threading.Event()
        super(SubThread, self).__init__()

    def run(self):
        while not self.should_stop.wait(1):
            #time.sleep(1) # No need to sleep, since we're waiting for 1 second above.
            self.source.somevar["anotherkey"].append(5)
            print "My subthread : ", id(self.source), id(self.source.somevar), self.source.somevar

    def stop(self):
        """ Call this to abort the thread. """
        self.should_stop.set()

这并不像使用 process.terminate() 那样立即终止，因为您必须实际调用 should_stop.wait()在线程停止之前。

为了使SubProcess正常工作，您需要使用进程安全的共享变量。 multiprocessing 模块为此提供了 multiprocessing.Manager；它允许您在管理器进程中创建共享变量。不过，更新字典的方式实际上要正确处理有点棘手，因为更改从 Manager 返回的 Proxy 对象内部的可变值存在一些限制。 >(请参阅 here 上的注释)。您必须显式地将更新后的列表重新分配给字典才能正确更新字典:

class SubProcess(multiprocessing.Process):

    def __init__(self, source):
        self.source = source
        super(SubProcess, self).__init__()

    def run(self):
        while 1:
            time.sleep(1) # Waiting for request...
            # Can't do this with a Manager.dict
            #self.source.somevar["anotherkey"].append(5)

            # Do this instead. You'd need to do it with SubThread.run, too.
            l = self.source.somevar["anotherkey"]
            l.append(5)
            self.source.somevar["anotherkey"] = l
            print "My subprocess : ", id(self.source), id(self.source.somevar), self.source.somevar

class Source:

    def __init__(self):
        self.m = multiprocessing.Manager()
        # somevar is now process-safe.
        self.somevar = self.m.dict({"akey": "THE key", "anotherkey": [5]})

    # The rest is the same