연속 프로세스 출력을 캡처할 때 Python 프로그램이 중단되는 것을 어떻게 방지할 수 있습니까?-파이썬 튜토리얼-php.cn

연속 프로세스 출력을 캡처할 때 Python 프로그램이 중단되는 것을 어떻게 방지할 수 있습니까?

Susan Sarandon

풀어 주다： 2024-10-29 18:19:09

원래의

923명이 탐색했습니다.

How Can I Prevent Python Programs from Hanging When Capturing Continuous Process Output?

Python에서 중단 없이 프로세스 출력 읽기 중지

os.popen()을 사용하여 프로세스 출력을 캡처할 때 다음과 같은 경우 프로그램이 중단될 수 있습니다. 프로세스는 지속적으로 데이터를 출력합니다. 이 문제를 해결하려면 다음과 같은 대체 방법을 사용하는 것이 좋습니다.

하위 프로세스 및 스레드 사용

subprocess.Popen으로 프로세스를 시작하고, 출력을 읽는 스레드를 생성하고, 그리고 이를 큐에 저장합니다. 특정 시간 초과 후 프로세스를 종료합니다.

<code class="python">import subprocess
import threading
import time

def read_output(process, append):
    for line in iter(process.stdout.readline, ""):
        append(line)

def main():
    process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
    try:
        q = collections.deque(maxlen=200)
        t = threading.Thread(target=read_output, args=(process, q.append))
        t.daemon = True
        t.start()
        time.sleep(2)
    finally:
        process.terminate()
    print(''.join(q))</code>

로그인 후 복사

신호 처리기 사용

지정된 시간 초과 후 예외를 발생시키려면 signal.alarm()을 사용하세요. 이렇게 하면 프로세스가 강제 종료되어 출력이 캡처될 수 있습니다.

<code class="python">import signal
import subprocess
import time

def alarm_handler(signum, frame):
    raise Alarm

def main():
    process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
    signal.signal(signal.SIGALRM, alarm_handler)
    signal.alarm(2)
    q = collections.deque(maxlen=200)
    try:
        for line in iter(process.stdout.readline, ""):
            q.append(line)
        signal.alarm(0)
    except Alarm:
        process.terminate()
    finally:
        print(''.join(q))</code>

로그인 후 복사

타이머 사용

threading.Timer를 사용하여 시간 초과 후 프로세스를 종료합니다.

<code class="python">import threading
import subprocess

def main():
    process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
    timer = threading.Timer(2, process.terminate)
    timer.start()
    q = collections.deque(process.stdout, maxlen=200)
    timer.cancel()
    print(''.join(q))</code>

로그인 후 복사

대체 접근법