Comment éviter que les programmes Python ne se bloquent lors de la lecture de la sortie du processus ?

Arrêter de lire la sortie du processus en Python sans blocage ?

Problème :

Un programme Python doit interagir avec un programme externe processus (par exemple, « top ») qui produit en continu une sortie. Cependant, la simple lecture directe de la sortie peut provoquer le blocage du programme indéfiniment.

Solution :

Pour éviter le blocage, il est essentiel d'utiliser des mécanismes non bloquants ou asynchrones lorsque sortie du processus de lecture. Voici quelques approches possibles :

Fichier temporaire spoolé (recommandé)

Cette méthode utilise un objet fichier dédié pour stocker la sortie du processus.

# ! /usr/bin/env python<br>import subprocess<br>import tempfile<br>import time</p>
<p>def main():</p>
<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()

if nom == "__main__":

main()

Lecture de sortie basée sur les threads

Cette approche utilise un thread séparé pour lire en continu la sortie du processus tout en le thread principal poursuit d'autres tâches.

import collections<br>import subprocess<br>import threading<br>import time</p>
<p>def read_output(process, append):</p>
<pre class="brush:php;toolbar:false">for line in iter(process.stdout.readline, ""):
    append(line)

def main():

# 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))

if nom == "__main__":

main()

signal.alarm() (Unix uniquement)

Cette méthode utilise des signaux Unix pour terminer le processus après un délai d'attente spécifié, que toutes les sorties aient été lues ou non.

importer des collections<br>signal d'importation<br>sous-processus d'importation</p>
<p>class Alarm(Exception):</p>
<pre class="brush:php;toolbar:false">pass

def alarm_handler(signum, frame):

raise Alarm

def main():

# 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))

if nom == "__main__":

main()

threading.Timer

Cette approche utilise une minuterie pour terminer le processus après un délai d'attente spécifié. Il fonctionne à la fois sur les systèmes Unix et Windows.

import collections<br>import subprocess<br>import threading</p>
<p>def main():</p>
<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))

if name == "__main__":

main()

Aucun fil de discussion, aucun signal

Cette méthode utilise un boucle temporelle simple pour vérifier la sortie du processus et la tuer si elle dépasse un délai d'attente spécifié.

import collections<br>import subprocess<br>import sys<br>import time</p> <p>def main():</p>
<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))

if nom == "__main__":

main()

Remarque : Le nombre de lignes stockées peut être ajusté selon les besoins en définissant le paramètre 'maxlen' de la structure de données deque.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!