可停止多线程中的 multiprocessing

Question

我正在尝试使用多线程后的多处理。我需要用新线程外给出的命令停止所有例程。下面的代码有一个使用线程属性作为停止标志来停止线程的工作示例。然而，当它与多处理结合使用时，似乎使用线程属性不起作用。

import threading
import multiprocessing
import time
def bar(param):
current_thread = threading.currentThread()
print("Current thread: ", current_thread)
time.sleep(0.01)
if getattr(current_thread, "do_run", True):
print("value:", param)
bar(param+1)
else:
print("Stop")
return param

def foo():
#without multiprocessing
bar(10)

def foo2():
#with multi processing
p = multiprocessing.Process(target=bar, args=(10,))
p.start()
p.join() 

if __name__ == '__main__':   
newThread = threading.Thread(target=foo)
newThread.start()
time.sleep(0.1)
newThread.do_run = False

newThread2 = threading.Thread(target=foo2)
newThread2.start()
time.sleep(0.1)
newThread2.do_run = False

输出:

Current thread:  <Thread(Thread-1, started 6004)>
value: 10
Current thread:  <Thread(Thread-1, started 6004)>
value: 11
Current thread:  <Thread(Thread-1, started 6004)>
value: 12
Current thread:  <Thread(Thread-1, started 6004)>
value: 13
Current thread:  <Thread(Thread-1, started 6004)>
value: 14
Current thread:  <Thread(Thread-1, started 6004)>
value: 15
Current thread:  <Thread(Thread-1, started 6004)>
Stop
Current thread:  <_MainThread(MainThread, started 4624)>
value: 10
Current thread:  <_MainThread(MainThread, started 4624)>
value: 11
.
.
.
#does not stop

Answer 1

当您使用p = multiprocessing.Process(target=bar, args=(10,))创建一个新进程时，该进程将有自己的主线程，bar将在其上运行，而该主线程与主进程的主线程设置&;do_run&;到False。这就是你问题的症结所在。

最简单的修复方法是将threading.Event实例或multiprocessing.Event实例传递给bar，这取决于它是在主进程的另一个线程中运行还是在子进程的另一个线程中运行。然后，当事件被"设置"时，它就会停止。我还将修改bar，将其尾部递归替换为while循环，并将print语句合并为清晰的语句:

import threading
import multiprocessing
import time
def bar(param, event):
while True:
current_thread = threading.currentThread()
print("Current thread: ", current_thread, end = ' ')
time.sleep(0.01)
if not event.is_set():
print("value:", param, flush=True)
param += 1
else:
print("Stop", flush=True)
return param
def foo(event):
#without multiprocessing
bar(10, event)
def foo2(event):
#with multi processing
p = multiprocessing.Process(target=bar, args=(10, event))
p.start()
time.sleep(0.1)
event.set()
p.join()
if __name__ == '__main__':
event1 = threading.Event()
newThread = threading.Thread(target=foo, args=(event1,))
newThread.start()
event2 = multiprocessing.Event()
newThread2 = threading.Thread(target=foo2, args=(event2,))
newThread2.start()
# for the "without multiprocessing" case:
time.sleep(0.1)
event1.set()
newThread.join()
newThread2.join()

打印:

Current thread:  <Thread(Thread-1, started 31712)> value: 10
Current thread:  <Thread(Thread-1, started 31712)> value: 11
Current thread:  <Thread(Thread-1, started 31712)> value: 12
Current thread:  <Thread(Thread-1, started 31712)> value: 13
Current thread:  <Thread(Thread-1, started 31712)> value: 14
Current thread:  <Thread(Thread-1, started 31712)> value: 15
Current thread:  <Thread(Thread-1, started 31712)> value: 16
Current thread:  <Thread(Thread-1, started 31712)> value: 17
Current thread:  <Thread(Thread-1, started 31712)> value: 18
Current thread:  <Thread(Thread-1, started 31712)> value: 19
Current thread:  <Thread(Thread-1, started 31712)> Stop
Current thread:  <_MainThread(MainThread, started 25948)> value: 10
Current thread:  <_MainThread(MainThread, started 25948)> Stop

Answer 2

这是一个使用@martineau建议的解决方案

import threading
import multiprocessing
import time
def bar(param):
current_thread = threading.currentThread()
print("Current thread: ", current_thread)
time.sleep(0.3)
if getattr(current_thread, "do_run", True):
print("value:", param)
bar(param+1)
else:
print("Stop")
return param

def foo2():
#with multi processing
global p
p = multiprocessing.Process(target=bar, args=(10,))
p.start()
p.join() 

if __name__ == '__main__':   
global p
newThread2 = threading.Thread(target=foo2)
newThread2.start()
time.sleep(1)
p.terminate()

这是@Booboo的编辑解决方案

import threading
import multiprocessing
import time
def bar(param, event):
while True:
current_thread = threading.currentThread()
print("Current thread: ", current_thread, end = ' ')
time.sleep(0.1)
if not event.is_set():
print("value:", param, flush=True)
param += 1
else:
print("Stop", flush=True)
return param
def foo2(event):
#with multi processing
p = multiprocessing.Process(target=bar, args=(10, event))
p.start()
if __name__ == '__main__':
event2 = multiprocessing.Event()
newThread2 = threading.Thread(target=foo2, args=(event2,))
newThread2.start()
time.sleep(1)
event2.set()
newThread2.join()