python - 从 gevent-subprocess 获取实时标准输出？

Question

我正试图通过 POPEN 获得一个进程的标准输出，只要它在那里。使用 gevent 1.0 readline() 和 read() 仍然会阻塞进程并等待进程完成。有什么线索吗？是的，我到处寻找一个简单的解决方案。没有线程必须是可能的，对吧？

Answer 1

import gevent
from gevent.subprocess import Popen, PIPE

def cron():
    while True:
        print("cron")
        gevent.sleep(0.5)

g = gevent.spawn(cron)
def subp():
    sub = Popen('sleep 1; ping www.google.com -c 2; sleep 5; uname', stdout=PIPE, shell=True)
    while True:
        s = sub.stdout.readline()
        if s == "":
            break
        else:
            print s.strip()
    g.kill()
subp()

也可以查看此要点以获取名为 prun 的工具：https ://gist.github.com/zhangchunlin/05576572b628f5bf9d74

Answer 2

我用一个小技巧解决了这个问题。每当将一行写入标准输出时，只需使用 self.stream.flush() 和一个辅助类刷新 Python 的行缓冲区。

Answer 3

这是一种可用于同时读取stdout和 stderr 的技术。它假设您可以提供一个缓冲区来写入从流中读取的数据，但是正如您所看到的，它可以很容易地通过调用日志记录模块、简单的打印语句或回调来即时执行某些操作来替换，或者：

import gevent
import gevent.subprocess

def read_stream(stream, buf):

    try:
        while not stream.closed:
            l = stream.readline()
            if not l: break
            buf.write(l)
    except RuntimeError:
        # process was terminated abruptly
        pass


p = gevent.subprocess.Popen(...)
stdout = ... #create buffer with write() method
stderr = ... #create buffer with write() method

gevent.spawn(read_stream, p.stdout, stdout)
gevent.spawn(read_stream, p.stderr, stderr)

status = p.wait()

编辑：根据 JF Sebastien 的评论，我实现了一个完整的示例，该示例演示了使用 greenlets 对输出进行并发实时读出。我没有像上面建议的那样累积输出，只是为了这个例子而打印它。这里是：

import gevent
import gevent.subprocess

def count_greenlets(s):
    '''See: http://stackoverflow.com/a/20027162/712525'''

    import gc
    from greenlet import greenlet
    greenlets = [obj for obj in gc.get_objects() if isinstance(obj, greenlet)]
    print('At "%s", greenlets: %d' % (s, len(greenlets)))
    for k in greenlets:
        print('  * %s' % (k,))


def read_stream(stream):

    try:
        while not stream.closed:
            l = stream.readline()
            if not l: break
            print(l.rstrip())
    except RuntimeError:
        # process was terminated abruptly
        pass


count_greenlets('start')

p1 = gevent.subprocess.Popen('ping -c 5 www.google.com', stdout=gevent.subprocess.PIPE, stderr=gevent.subprocess.PIPE, shell=True)
gevent.spawn(read_stream, p1.stdout)
gevent.spawn(read_stream, p1.stderr)

count_greenlets('after p1')

p2 = gevent.subprocess.Popen('ping -c 5 www.facebook.com', stdout=gevent.subprocess.PIPE, stderr=gevent.subprocess.PIPE, shell=True)
gevent.spawn(read_stream, p2.stderr)
gevent.spawn(read_stream, p2.stdout)

count_greenlets('after p2')

p1.wait()
count_greenlets('after p1 wait')
p2.wait()
count_greenlets('after p2 wait')

count_greenlets('end')

它提供以下输出：

At "start", greenlets: 1
  * <greenlet.greenlet object at 0x1060d0690>
At "after p1", greenlets: 4
  * <Hub at 0x106300f50 select default pending=0>
  * <Greenlet at 0x10646b0f0: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <Greenlet at 0x10646b190: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <greenlet.greenlet object at 0x1060d0690>
At "after p2", greenlets: 6
  * <Hub at 0x106300f50 select default pending=0>
  * <Greenlet at 0x10646b0f0: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <Greenlet at 0x10646b190: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <Greenlet at 0x10646b230: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <Greenlet at 0x10646b2d0: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <greenlet.greenlet object at 0x1060d0690>
PING www.google.com (172.217.19.164): 56 data bytes
64 bytes from 172.217.19.164: icmp_seq=0 ttl=56 time=12.722 ms
PING star-mini.c10r.facebook.com (31.13.91.36): 56 data bytes
64 bytes from 31.13.91.36: icmp_seq=0 ttl=87 time=29.673 ms
64 bytes from 172.217.19.164: icmp_seq=1 ttl=56 time=11.863 ms
64 bytes from 31.13.91.36: icmp_seq=1 ttl=87 time=31.389 ms
64 bytes from 172.217.19.164: icmp_seq=2 ttl=56 time=13.492 ms
64 bytes from 31.13.91.36: icmp_seq=2 ttl=87 time=29.921 ms
64 bytes from 172.217.19.164: icmp_seq=3 ttl=56 time=12.488 ms
64 bytes from 31.13.91.36: icmp_seq=3 ttl=87 time=30.859 ms
64 bytes from 172.217.19.164: icmp_seq=4 ttl=56 time=13.053 ms

--- www.google.com ping statistics ---
5 packets transmitted, 5 packets received, 0.0% packet loss
round-trip min/avg/max/stddev = 11.863/12.724/13.492/0.547 ms
At "after p1 wait", greenlets: 4
  * <Hub at 0x106300f50 select default pending=0>
  * <Greenlet at 0x10646b230: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <Greenlet at 0x10646b2d0: read_stream(<gevent._fileobjectposix.FileObjectPosix object at)>
  * <greenlet.greenlet object at 0x1060d0690>
64 bytes from 31.13.91.36: icmp_seq=4 ttl=87 time=30.379 ms

--- star-mini.c10r.facebook.com ping statistics ---
5 packets transmitted, 5 packets received, 0.0% packet loss
round-trip min/avg/max/stddev = 29.673/30.444/31.389/0.622 ms
At "after p2 wait", greenlets: 2
  * <Hub at 0x106300f50 select default pending=0>
  * <greenlet.greenlet object at 0x1060d0690>
At "end", greenlets: 2
  * <Hub at 0x106300f50 select default pending=0>
  * <greenlet.greenlet object at 0x1060d0690>