스레드를 종료해야하지만 읽기 / 입력을 기다리기 때문에 정기적으로 플래그를 확인할 수 없습니다.
간단한 예 :
import threading, time
class Test(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
print(input("wainting for input: "))
th = Test()
th.start()
time.sleep(5)
print("killing!")
th.join(5)
print(th.is_alive())
보다 실제적인 예는 다음과 같습니다 (정지시 스레드 종료-더 이상 출력되지 않음).
import threading, time
class Test(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def call(args):
return subprocess.Popen(" ".join(args), shell=True, stderr=subprocess.PIPE)
def run(self):
mainProcess = call([ any program that could hang])
out = None
while mainProcess.returncode != 0 or out == '' and mainProcess.poll() != None:
out = mainProcess.stderr.read(1)
if out != '':
sys.stdout.write(out)
sys.stdout.flush()
th = Test()
th.start()
time.sleep(5)
print("killing!")
th.join(5)
print(th.is_alive())
더 나은 접근법이 있다면 저도 기뻐할 것입니다.
다음은 다음을 사용하여 교수형 프로세스 문제를 해결하는 방법의 예입니다 select.
import threading
import select
import subprocess
import sys
def watch_output(args, timeout):
process = subprocess.Popen(args, stdout=subprocess.PIPE)
while True:
ready_to_read, _, _ = select.select([process.stdout], [], [], timeout)
if not ready_to_read:
print "hanging process"
process.kill()
break
out = ready_to_read[0].read(1)
if not out:
print "normal exit"
break
sys.stdout.write(out)
sys.stdout.flush()
return process.wait()
watch_output(['ls'], timeout=10)
또는 시간 제한이있는 입력도 가능합니다.
def read_input(prompt, timeout):
sys.stdout.write(prompt)
sys.stdout.flush()
ready_to_read, _, _ = select.select([sys.stdin], [], [], timeout)
if not ready_to_read:
return None
return ready_to_read[0].readline()
print read_input("wainting for input (4s): ", 4)
이 기사는 인터넷에서 수집됩니다. 재 인쇄 할 때 출처를 알려주십시오.
침해가 발생한 경우 연락 주시기 바랍니다[email protected] 삭제
몇 마디 만하겠습니다