Wie kann verhindert werden, dass Python-Programme beim Lesen der Prozessausgabe hängen bleiben?-Python-Tutorial-php.cn

Wie kann verhindert werden, dass Python-Programme beim Lesen der Prozessausgabe hängen bleiben?

Susan Sarandon

Freigeben： 2024-11-02 13:54:29

Original

527 Leute haben es durchsucht

How to Avoid Python Programs from Hanging When Reading Process Output?

Prozessausgabe in Python nicht mehr lesen, ohne zu hängen?

Problem:

Ein Python-Programm muss mit einem externen Programm interagieren Prozess (z. B. „oben“), der kontinuierlich Output produziert. Das einfache direkte Lesen der Ausgabe kann jedoch dazu führen, dass das Programm auf unbestimmte Zeit hängen bleibt.

Lösung:

Um ein Hängenbleiben zu verhindern, ist es wichtig, nicht blockierende oder asynchrone Mechanismen zu verwenden, wenn Ausgabe des Lesevorgangs. Hier sind einige mögliche Ansätze:

Temporäre Spooldatei (empfohlen)

Diese Methode verwendet ein dediziertes Dateiobjekt zum Speichern der Prozessausgabe.

#! /usr/bin/env python<br>Unterprozess importieren<br>Tempfile importieren<br>Zeitpunkt importieren</p>
<p>def main():</p>
<div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false"># Open a temporary file (automatically deleted on closure)
f = tempfile.TemporaryFile()

# Start the process and redirect stdout to the file
p = subprocess.Popen(["top"], stdout=f)

# Wait for a specified duration
time.sleep(2)

# Kill the process
p.terminate()
p.wait()

# Rewind and read the captured output from the file
f.seek(0)
output = f.read()

# Print the output
print(output)
f.close()

Nach dem Login kopieren

Dieser Ansatz verwendet einen separaten Thread, um die Prozessausgabe kontinuierlich zu lesen Der Hauptthread fährt mit anderen Aufgaben fort.

Sammlungen importieren Unterprozess importieren Threading importieren Importzeit def read_output(process, append): <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false">for line in iter(process.stdout.readline, ""): append(line)

# Start the process and redirect stdout process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True) # Create a thread for output reading q = collections.deque(maxlen=200) t = threading.Thread(target=read_output, args=(process, q.append)) t.daemon = True t.start() # Wait for the specified duration time.sleep(2) # Print the saved output print(''.join(q))

Sammlungen importieren Signal importieren Unterprozess importieren Klasse Alarm (Ausnahme): <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false">pass

# Start the process and redirect stdout process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True) # Set signal handler signal.signal(signal.SIGALRM, alarm_handler) signal.alarm(2) try: # Read and save a specified number of lines q = collections.deque(maxlen=200) for line in iter(process.stdout.readline, ""): q.append(line) signal.alarm(0) # Cancel alarm except Alarm: process.terminate() finally: # Print the saved output print(''.join(q))

Sammlungen importieren Unterprozess importieren Threading importieren def main(): <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false"># Start the process and redirect stdout process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True) # Create a timer for process termination timer = threading.Timer(2, process.terminate) timer.start() # Read and save a specified number of lines q = collections.deque(maxlen=200) for line in iter(process.stdout.readline, ""): q.append(line) timer.cancel() # Print the saved output print(''.join(q))

Sammlungen importieren Unterprozess importieren Systeme importieren Importzeit def main(): <div class="code" style="position:relative; padding:0px; margin:0px;"><pre class="brush:php;toolbar:false">args = sys.argv[1:] if not args: args = ['top'] # Start the process and redirect stdout process = subprocess.Popen(args, stdout=subprocess.PIPE, close_fds=True) # Save a specified number of lines q = collections.deque(maxlen=200) # Set a timeout duration timeout = 2 now = start = time.time() while (now - start) < timeout: line = process.stdout.readline() if not line: break q.append(line) now = time.time() else: # On timeout process.terminate() # Print the saved output print(''.join(q))