<p><code>FDRedirector</code>的问题是,它只从管道读取一次。由于管道的生产者端和使用者端都处于同一进程中,一旦缓冲区已满,这将阻塞管道的写入端。解决方法是从读取端连续读取,而不阻塞另一端。一种方法是生成一个消费者线程。在</p>
<p>这里有一段代码可以做到这一点:将stderr重定向到管道并使用线程连续读取。在</p>
<p><strong>更新</strong>:正如OP的评论所建议的,代码现在是针对python3的。假设您希望捕获的输出是python3字符串(即Unicode字符串),现在,<code>redirect</code>将读取的字节转换为字符串。为此,它接受<code>encoding</code>和<code>errors</code>参数——就像Python的<code>decode</code>(并使用相同的默认值)。这解决了一般用例的一个实际问题:如果您想将捕获的数据作为一个字符串进行进一步处理,那么您必须知道stderr是以哪种编码方式写入的。对于另一个答案中的方法也是如此,其中流被重定向到一个文件。{6}你可以用cd6}来修改代码。在</p>
<pre><code>import os, sys, threading
from contextlib import contextmanager
from io import StringIO
STDOUT, STDERR = 1, 2
@contextmanager
def redirect(fd, encoding="utf-8", errors="strict"):
# Save original handle so we can restore it later.
saved_handle = os.dup(fd)
# Redirect `fd` to the write end of the pipe.
pipe_read, pipe_write = os.pipe()
os.dup2(pipe_write, fd)
os.close(pipe_write)
# This thread reads from the read end of the pipe.
def consumer_thread(f, data):
while True:
buf = os.read(f, 1024)
if not buf: break
data.write(buf.decode(encoding, errors))
os.close(f)
return
# Spawn consumer thread, and give it a mutable `data` item to
# store the redirected output.
data = StringIO()
thread = threading.Thread(target = consumer_thread, args=(pipe_read, data))
thread.start()
yield data
# Cleanup: flush streams, restore `fd`
{ STDERR: sys.stderr, STDOUT: sys.stdout}[fd].flush()
os.dup2(saved_handle, fd)
os.close(saved_handle)
thread.join()
</code></pre>
<p>然后,<code>redirect</code>上下文管理器可用于临时将任何流重定向到内部缓冲区。注意,在Jupyter下,<code>sys.stderr</code>是连接到<code>STDERR</code>的<em>而不是</em>,它是一个<code>ipykernel.iostream.OutStream</code>对象,模拟Python<code>file</code>。所以我们必须<code>os.write</code>到{<cd9>}才能看到重定向工作。在</p>
^{pr2}$