연속 프로세스 출력을 읽을 때 Python 프로그램이 중단되는 것을 방지하는 방법은 무엇입니까?

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

배경

연속 출력을 생성하는 도구와 함께 Python의 os.popen() 함수를 사용할 때, 출력을 읽으려고 할 때 프로그램이 종종 중단됩니다.

os.popen() 문제

문제가 있는 라인 process = os.popen("top").readlines()는 다음으로 인해 프로그램을 중단시킵니다. 전체 프로세스 출력을 한 번에 읽으려고 시도하는 readlines().

subprocess.Popen()을 사용한 솔루션

이 문제를 해결하려면 os.popen 대신 subprocess.Popen()을 사용하세요. (). 수정된 예는 다음과 같습니다.

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

# Start "top" process with stdout redirection
process = subprocess.Popen(["top"], stdout=subprocess.PIPE)

# Wait for 2 seconds
time.sleep(2)

# Send kill signal to "top" process
os.popen("killall top")

# Read process output
output, _ = process.communicate()
print(output.decode())</code>

수정된 코드:

• readlines() 대신에 communications()를 사용하여 변수에 프로세스 출력을 캡처합니다.
• "최상위" 프로세스에 종료 신호를 보냅니다.
• 프로세스의 I/O 스트림에 대한 파일 끝을 선언하고 프로그램을 종료합니다.

꼬리형 접근 방식

프로세스 출력의 일부만 필요한 경우 꼬리 모양 솔루션을 사용하여 특정 라인 수를 캡처할 수 있습니다.

스레드 기반 접근 방식

프로세스 캡처 별도의 스레드로 출력하려면 다음을 시도해 보세요.

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

# Start process with stdout redirection
process = subprocess.Popen(["top"], stdout=subprocess.PIPE)

# Define function to read process output in a thread
def read_output(process):
    for line in iter(process.stdout.readline, ""):
        ...  # Implement your logic here to process each line

# Create and start a thread for reading and processing output
reading_thread = threading.Thread(target=read_output, args=(process,))
reading_thread.start()

# Wait for 2 seconds, then terminate the process
time.sleep(2)
process.terminate()

# Wait for the reading thread to complete
reading_thread.join()</code>

signal.alarm() 접근 방식

또한 signal.alarm()을 사용하여 지정된 시간 초과 후 프로세스를 종료할 수도 있습니다.

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

# Define signal handler
def alarm_handler(signum, frame):
    # Raise an exception to terminate the process reading
    raise Exception

# Set signal handler and alarm for 2 seconds
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(2)

# Start process with stdout redirection
process = subprocess.Popen(["top"], stdout=subprocess.PIPE)

# Capture process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines)
for line in iter(process.stdout.readline, ""):
    q.append(line)

# Cancel alarm
signal.alarm(0)

# Print captured output
print(''.join(q))</code>

threading.Timer 접근 방식

또는 threading.Timer를 사용하여 프로세스 종료를 예약할 수 있습니다.

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

# Define function to terminate the process
def terminate_process(process):
    process.terminate()

# Start process with stdout redirection
process = subprocess.Popen(["top"], stdout=subprocess.PIPE)

# Create and start a timer to terminate process in 2 seconds
timer = threading.Timer(2, terminate_process, [process])
timer.start()

# Capture process output
number_of_lines = 200
q = collections.deque(process.stdout, maxlen=number_of_lines)

# Cancel timer
timer.cancel()

# Print captured output
print(''.join(q))</code>

위 내용은 연속 프로세스 출력을 읽을 때 Python 프로그램이 중단되는 것을 방지하는 방법은 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!