我希望能够抑制特定线程内的任何打印到标准输出。这是我尝试过的:
import sys, io, time
from threading import Thread
def do_thread_action():
# Disable stdout
sys.stdout = io.StringIO()
print("don't print this 1")
time.sleep(1)
print("don't print this 2")
time.sleep(1)
print("don't print this 3")
# Re-enable stdout
sys.stdout = sys.__stdout__
thread = Thread(target=do_thread_action)
thread.start()
time.sleep(1.5)
# Print this to stdout
print('Print this')
thread.join()
但是这不起作用,因为
sys.stdout
对于 thread
和主线程来说都是全局的。
如何抑制线程内
do_thread_action
内部的打印,但不抑制线程外部的打印?
您可以通过其
is_alive
属性或(可能更好)通过 Event
监视线程的状态,并在线程工作完成后恢复 sys.stdout
thread.is_alive()
import sys, time
from threading import Thread
# store the original definition of sys.stdout
og_stdout = sys.stdout
def do_thread_action() -> None:
# disable stdout
sys.stdout = None
print("don't print this 1")
time.sleep(1)
print("don't print this 2")
time.sleep(1)
print("don't print this 3")
stop_event.set()
thread = Thread(target=do_thread_action)
thread.start()
# poll thread state
while thread.is_alive():
time.sleep(0.5) # sleep duration is arbitrary here...
else: # restore stdout once the thread is dead
sys.stdout = og_stdout
# Print this to stdout
print('Print this')
thread.join()
Event
import sys, time
from threading import Thread, Event
# store the original definition of sys.stdout
og_stdout = sys.stdout
# instantiate a Threading.Event object
stop_event = Event()
def do_thread_action(stop_event: Event) -> None:
# disable stdout
sys.stdout = None
print("don't print this 1")
time.sleep(1)
print("don't print this 2")
time.sleep(1)
print("don't print this 3")
stop_event.set() # set the event flag when the work is done
thread = Thread(target=do_thread_action, args=(stop_event,)) # pass in the stop_event
thread.start()
# poll thread event
while not stop_event.is_set():
time.sleep(0.5) # sleep duration is arbitrary here...
else: # restore stdout once the thread is dead
sys.stdout = og_stdout
# Print this to stdout
print('Print this')
thread.join()
也就是说,我倾向于不喜欢依赖这样的民意调查,因为它确实涉及到一些人坐在
while
循环中。