如何使用 Python 将标准输入/标准输出通过管道传输到 Perl 脚本答案

【问题标题】：How can I use Python to pipe stdin/stdout to Perl script如何使用 Python 将标准输入/标准输出通过管道传输到 Perl 脚本
【发布时间】：2012-01-02 07:07:28
【问题描述】：

这个 Python 代码通过 Perl 脚本很好地管道数据。

import subprocess
kw = {}
kw['executable'] = None
kw['shell'] = True
kw['stdin'] = None
kw['stdout'] = subprocess.PIPE
kw['stderr'] = subprocess.PIPE
args = ' '.join(['/usr/bin/perl','-w','/path/script.perl','<','/path/mydata'])
subproc = subprocess.Popen(args,**kw)
for line in iter(subproc.stdout.readline, ''):
    print line.rstrip().decode('UTF-8')

但是，它要求我首先将缓冲区保存到磁盘文件 (/path/mydata)。在 Python 代码中循环遍历数据并逐行传递给子进程会更简洁，如下所示：

import subprocess
kw = {}
kw['executable'] = '/usr/bin/perl'
kw['shell'] = False
kw['stderr'] = subprocess.PIPE
kw['stdin'] = subprocess.PIPE
kw['stdout'] = subprocess.PIPE
args = ['-w','/path/script.perl',]
subproc = subprocess.Popen(args,**kw)
f = codecs.open('/path/mydata','r','UTF-8')
for line in f:
    subproc.stdin.write('%s\n'%(line.strip().encode('UTF-8')))
    print line.strip()  ### code hangs after printing this ###
    for line in iter(subproc.stdout.readline, ''):
        print line.rstrip().decode('UTF-8')
subproc.terminate()
f.close()

将第一行发送到子进程后，代码与 readline 一起挂起。我有其他可执行文件完美地使用了完全相同的代码。

我的数据文件可能非常大 (1.5 GB) 有没有办法在不保存到文件的情况下完成数据管道传输？为了与其他系统兼容，我不想重写 perl 脚本。

【问题讨论】：

标签： python pipe

【解决方案1】：

您的代码在该行阻塞：

for line in iter(subproc.stdout.readline, ''):

因为此迭代可以终止的唯一方法是到达 EOF（文件结尾）时，这将在子进程终止时发生。您不想等到进程终止，但是，您只想等到它完成处理发送给它的行。

此外，正如 Chris Morgan 已经指出的那样，您遇到了缓冲问题。另一个question on stackoverflow 讨论了如何使用子进程进行非阻塞读取。我已经将那个问题的代码快速而肮脏地改编为你的问题：

def enqueue_output(out, queue):
    for line in iter(out.readline, ''):
        queue.put(line)
    out.close()

kw = {}
kw['executable'] = '/usr/bin/perl'
kw['shell'] = False
kw['stderr'] = subprocess.PIPE
kw['stdin'] = subprocess.PIPE
kw['stdout'] = subprocess.PIPE
args = ['-w','/path/script.perl',]
subproc = subprocess.Popen(args, **kw)
f = codecs.open('/path/mydata','r','UTF-8')
q = Queue.Queue()
t = threading.Thread(target = enqueue_output, args = (subproc.stdout, q))
t.daemon = True
t.start()
for line in f:
    subproc.stdin.write('%s\n'%(line.strip().encode('UTF-8')))
    print "Sent:", line.strip()  ### code hangs after printing this ###
    try:
        line = q.get_nowait()
    except Queue.Empty:
        pass
    else:
        print "Received:", line.rstrip().decode('UTF-8')

subproc.terminate()
f.close()

您很可能需要修改此代码，但至少不会阻塞。

【讨论】：

【解决方案2】：

感谢 srgerg。我也尝试过线程解决方案。然而，仅此解决方案总是挂起。我之前的代码和 srgerg 的代码都缺少最终解决方案，您的提示给了我最后一个想法。

最终解决方案写入足够多的虚拟数据，强制缓冲区中的最终有效行。为了支持这一点，我添加了跟踪有多少有效行写入标准输入的代码。线程循环打开输出文件，保存数据，并在读取行等于有效输入行时中断。此解决方案确保它可以逐行读取和写入任何大小的文件。

def std_output(stdout,outfile=''):
    out = 0
    f = codecs.open(outfile,'w','UTF-8')
    for line in iter(stdout.readline, ''):
        f.write('%s\n'%(line.rstrip().decode('UTF-8')))
        out += 1
        if i == out: break
    stdout.close()
    f.close()

outfile = '/path/myout'
infile = '/path/mydata'

subproc = subprocess.Popen(args,**kw)
t = threading.Thread(target=std_output,args=[subproc.stdout,outfile])
t.daemon = True
t.start()

i = 0
f = codecs.open(infile,'r','UTF-8')
for line in f:
    subproc.stdin.write('%s\n'%(line.strip().encode('UTF-8')))
    i += 1
subproc.stdin.write('%s\n'%(' '*4096)) ### push dummy data ###
f.close()
t.join()
subproc.terminate()

【讨论】：

【解决方案3】：

请参阅手册中提到的有关使用Popen.stdin 和Popen.stdout 的警告（就在Popen.stdin 上方）：

警告：使用 communicate() 而不是 .stdin.write、.stdout.read 或 .stderr.read 避免由于任何其他操作系统管道缓冲区填满并阻塞子进程而导致的死锁。

我意识到一次在内存中拥有一个千兆字节半的字符串并不是很理想，但使用communicate() 是一种将工作的方式，而你'我们观察到，一旦 OS 管道缓冲区填满，stdin.write() + stdout.read() 方式可能会陷入僵局。

您可以使用communicate() 吗？

【讨论】：