python: 没有外部命令的fork,并且分别捕获stdout和stderr

python: fork without an external command, and capturing stdout and stderr separately

我想在 python 中创建一个子进程,它没有 运行 外部命令...它只是 运行 一个定义的函数。我想分别捕获 stdoutstderr

我知道如何使用 os.fork()os.pipe(),但该机制只提供了两个 fd 供我使用。我正在寻找三个 fd:一个用于 stdin,一个用于 stdout,一个用于 stderr。在 运行 外部命令时使用 subprocess.Popen 很容易管理,但该函数似乎不允许分叉本地函数;只有一个单独的可执行文件。

在ruby中,popen3命令可以将“-”作为其命令参数,在这种情况下,会在不调用任何外部命令的情况下发生fork,并且3个fd的I提到的被退回。在 python 中是否有某种类似于此例程的 python?

  • 如果你想重定向 stdoutstderr 与子进程分开,你可以简单地为每个创建两个单独的管道,而不是一个。我已经分享了相关代码。

  • 您还可以阅读此主题以获取有关此主题的更多知识:Redirect stdout to a file in Python?

  • 我已经提到了从子进程写入 stdout 和 stderr 的两种方法 (Method1, Method2)

  • 如果你也想写入子进程的标准输入,你应该创建另一个文件描述符。这次 r 将转到子进程,而 w 将转到父进程。

import os
import sys
import time

# Create two pipes. One for sys.stdout, and one for sys.stderr
r_out, w_out = os.pipe()
r_err, w_err = os.pipe()

pid = os.fork()
if pid == 0:
    # Child process
    os.close(r_out)
    os.close(r_err)

    w1 = os.fdopen(w_out, "w")
    w2 = os.fdopen(w_err, "w")
    sys.stdout = w1
    sys.stderr = w2

    # Note that flush=True is necessary only if you want to ensure the order of messages printed
    # across method1, and method2 is maintained

    # Method 1: Standard Python print messages
    print("Redirected to stdout #2", flush=True)
    print("Redirected to stderr #2", file=sys.stderr, flush=True)

    # Method 2: Using system file descriptors
    stdout_fd = sys.stdout.fileno()
    os.write(stdout_fd, b'Redirected to stdout')

    stderr_fd = sys.stderr.fileno()
    os.write(stderr_fd, b'Redirected to stderr')

    # Restore original stdout, and stderr
    sys.stdout = sys.__stdout__
    sys.stderr = sys.__stderr__

    # Close the file descriptors
    w1.close()
    w2.close()

else:
    # Parent process
    os.close(w_out)
    os.close(w_err)

    r1 = os.fdopen(r_out)
    r2 = os.fdopen(r_err)
    for i in range(5):
        # Note that r1.read(), and r2.read() are non-blocking calls
        # You can run this while loop as long as you want.
        print("Read text (sysout):", r1.read())
        print("Read text (syserr):", r2.read())
        time.sleep(0.5)