連続プロセス出力を読み取るときに 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() の代わりに communication() を使用してプロセス出力を変数にキャプチャします。
• 「トップ」プロセスに kill シグナルを送信します。
• プロセスの 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 中国語 Web サイトの他の関連記事を参照してください。