从正在 Python 中写入的打开文件句柄中读取

Question

我知道这是流处理中的一个经典问题，但我不知道如何在Python中处理它。我有一个正在由活动进程写入的文件句柄。我想逐行使用该文件句柄中的内容，但我不想在等待读取时陷入死锁。我将继续阅读，直到 EOF 或循环阅读 60 秒，以先到者为准。有关如何执行此操作的建议将不胜感激。我对这个问题的伪代码描述如下。

proc = genprocess("command")
found_a = False
found_b = False
start = time.time()
while True:
    line = proc.readline()
    while line:
        if not found_a and grep(pattern_a, line):
            found_a = True
            print "Found A, now looking for B"
        elif not found_b and grep(pattern_b, line):
            found_b = True
            print "Found B, all done"
            break
    if time.time() - start > 60:
        break
    else:
        time.sleep(5)

proc.kill()

问题是这只能在每个时间间隔读取一行。相反，我希望循环内部尽可能多次地迭代，但不阻止等待新内容写入文件。一旦它读取了尽可能多的可用内容，它应该休眠 5 秒钟，以便积累更多内容。

Answer 1

如果你运行在 Unix 环境下，你可以使用 Python 的 select 模块 来等待用于文件句柄上的数据。另外，您可以使用Python的fcntl模块将文件句柄更改为非阻塞模式如此问题中所述。

例如，假设您的 proc 变量是支持 fileno() 的常规文件句柄：

file_num = proc.fileno()
old_flags = fcntl.fcntl(file_num, fcntl.F_GETFL)
fcntl.fcntl(file_num, fcntl.F_SETFL, old_flags | os.O_NONBLOCK)

Answer 2

上面链接的 fcntl 示例是可以的（除了它将进程置于繁忙的循环轮询中），但是我最终使用“select”来实现或多或少所需的功能。

    started = False
    while True:
        if (time.time() - start > wait_for) or started:
            break
        (rlist, wlist, xlist) = select([proc.stdout], [], [], wait_interval)
        if len(rlist) > 0:
            line = rlist[0].readline() # read one line (this blocks until '\n' is read)
        else: # nothing available to read from proc.stdout
            print ".",
            sys.stdout.flush()
            time.sleep(1)
            continue
        if re.search("daemon started", line):
            started = True

    if not started:
        proc.kill() # don't leave the process running if it didn't start properly

如果这是用户可能会按 CTRL-C 的情况，那么将整个事情放入 try/ except 块中并查找 KeyboardInterrupt 允许调用 proc.kill()，而不是让进程在后台运行。