#!/usr/bin/env python
"""Start process; wait 2 seconds; kill the process; print all process output."""
import subprocess
import tempfile
import time
def main():
# open temporary file (it automatically deleted when it is closed)
# `Popen` requires `f.fileno()` so `SpooledTemporaryFile` adds nothing here
f = tempfile.TemporaryFile()
# start process, redirect stdout
p = subprocess.Popen(["top"], stdout=f)
# wait 2 seconds
time.sleep(2)
# kill process
#NOTE: if it doesn't kill the process then `p.wait()` blocks forever
p.terminate()
p.wait() # wait for the process to terminate otherwise the output is garbled
# print saved output
f.seek(0) # rewind to the beginning of the file
print f.read(),
f.close()
if __name__=="__main__":
main()
类似尾部的解决方案，仅打印输出的一部分
您可以在另一个线程中读取进程输出，并将所需数量的最后几行保存在队列中：
import collections
import subprocess
import time
import threading
def read_output(process, append):
for line in iter(process.stdout.readline, ""):
append(line)
def main():
# start process, redirect stdout
process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
try:
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines) # atomic .append()
t = threading.Thread(target=read_output, args=(process, q.append))
t.daemon = True
t.start()
#
time.sleep(2)
finally:
process.terminate() #NOTE: it doesn't ensure the process termination
# print saved lines
print ''.join(q)
if __name__=="__main__":
main()
此变体需要q.append()成为原子操作。否则输出可能会损坏。
signal.alarm() http://docs.python.org/library/signal.html#signal.alarm解决方案
你可以使用signal.alarm() http://docs.python.org/library/signal.html#signal.alarm打电话给process.terminate()在指定的超时后而不是在另一个线程中读取。尽管它可能无法与subprocess模块。基于@亚历克斯·马泰利的回答 https://stackoverflow.com/questions/1191374/subprocess-with-timeout/1191537#1191537:
import collections
import signal
import subprocess
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
def main():
# start process, redirect stdout
process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
# set signal handler
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(2) # produce SIGALRM in 2 seconds
try:
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines)
for line in iter(process.stdout.readline, ""):
q.append(line)
signal.alarm(0) # cancel alarm
except Alarm:
process.terminate()
finally:
# print saved lines
print ''.join(q)
if __name__=="__main__":
main()
此方法仅适用于 *nix 系统。它可能会阻塞，如果process.stdout.readline()不回来。
threading.Timer http://docs.python.org/library/threading.html#threading.Timer解决方案
import collections
import subprocess
import threading
def main():
# start process, redirect stdout
process = subprocess.Popen(["top"], stdout=subprocess.PIPE, close_fds=True)
# terminate process in timeout seconds
timeout = 2 # seconds
timer = threading.Timer(timeout, process.terminate)
timer.start()
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(process.stdout, maxlen=number_of_lines)
timer.cancel()
# print saved lines
print ''.join(q),
if __name__=="__main__":
main()
这种方法也应该适用于 Windows。这里我用过process.stdout作为一个可迭代的；它可能会引入额外的输出缓冲，您可以切换到iter(process.stdout.readline, "")如果不希望采取这种方法。如果进程没有终止于process.terminate()然后脚本挂起。
无线程、无信号解决方案
import collections
import subprocess
import sys
import time
def main():
args = sys.argv[1:]
if not args:
args = ['top']
# start process, redirect stdout
process = subprocess.Popen(args, stdout=subprocess.PIPE, close_fds=True)
# save last `number_of_lines` lines of the process output
number_of_lines = 200
q = collections.deque(maxlen=number_of_lines)
timeout = 2 # seconds
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 saved lines
print ''.join(q),
if __name__=="__main__":
main()
此变体既不使用线程，也不使用信号，但会在终端中产生乱码输出。它将阻止如果process.stdout.readline() blocks.
