Python 安全地捕获来自多个子进程的实时输出

Question

https://stackoverflow.com/a/18422264/7238575 中解释了如何运行subprocess 并实时读出结果。但是，它看起来像是创建了一个名称为 test.log 的文件来执行此操作。这让我担心如果多个脚本在同一目录中使用此技巧，test.log 文件很可能已损坏。有没有一种不需要在 Python 之外创建文件的方法？或者我们可以确保每个进程使用一个唯一的日志文件吗？还是我完全误解了这种情况，不同程序同时写入同一个test.log文件没有风险？

Answer 1

您无需将实时输出写入文件。您可以使用 sys.stdout.write("your message") 将其简单地写入 STDOUT。

另一方面，您可以为每个进程生成唯一的日志文件：

import os
import psutil

pid = psutil.Process(os.getpid())
process_name = pid.name()
path, extension = os.path.splitext(os.path.join(os.getcwd(), "my_basic_log_file.log"))
created_log_file_name = "{0}_{1}{2}".format(path, process_name, extension)
print(created_log_file_name)

输出：

>>> python3 test_1.py
/home/my_user/test_folder/my_basic_log_file_python3.log

如果您看到上面的示例，我的进程名称是python3，所以这个进程名称被插入到“基本”日志文件名中。使用此解决方案，您可以为您的流程创建唯一的日志文件。

您可以使用setproctitle.setproctitle("my_process_name") 设置您的进程名称。 这是一个例子。

import os
import psutil
import setproctitle

setproctitle.setproctitle("milan_balazs")

pid = psutil.Process(os.getpid())
process_name = pid.name()
path, extension = os.path.splitext(os.path.join(os.getcwd(), "my_basic_log_file.log"))
created_log_file_name = "{0}_{1}{2}".format(path, process_name, extension)
print(created_log_file_name)

输出：

>>> python3 test_1.py
/home/my_user/test_folder/my_basic_log_file_milan_balazs.log

以前我写过一个相当复杂和安全的命令调用程序，它可以进行实时输出（而不是归档）。你可以检查一下：

import sys
import os
import subprocess
import select
import errno


def poll_command(process, realtime):
    """
     Watch for error or output from the process
    :param process: the process, running the command
    :param realtime: flag if realtime logging is needed
    :return: Return STDOUT and return code of the command processed
    """

    coutput = ""
    poller = select.poll()
    poller.register(process.stdout, select.POLLIN)

    fdhup = {process.stdout.fileno(): 0}
    while sum(fdhup.values()) < len(fdhup):
        try:
            r = poller.poll(1)
        except select.error as err:
            if not err.args[0] == errno.EINTR:
                raise
            r = []
        for fd, flags in r:
            if flags & (select.POLLIN | select.POLLPRI):
                c = version_conversion(fd, realtime)
                coutput += c
            else:
                fdhup[fd] = 1

    return coutput.strip(), process.poll()


def version_conversion(fd, realtime):
    """
    There are some differences between Python2/3 so this conversion is needed.
    """
    c = os.read(fd, 4096)
    if sys.version_info >= (3, 0):
        c = c.decode("ISO-8859-1")
    if realtime:
        sys.stdout.write(c)
        sys.stdout.flush()
    return c


def exec_shell(command, real_time_out=False):
    """
    Call commands.
    :param command: Command line.
    :param real_time_out: If this variable is True, the output of command is logging in real-time
    :return: Return STDOUT and return code of the command processed.
    """

    if not command:
        print("Command is not available.")
        return None, None

    print("Executing '{}'".format(command))
    rtoutput = real_time_out

    p = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    out, return_code = poll_command(p, rtoutput)

    if p.poll():
        error_msg = "Return code: {ret_code} Error message: {err_msg}".format(
            ret_code=return_code, err_msg=out
        )

        print(error_msg)

    print("[OK] - The command calling was successful. CMD: '{}'".format(command))

    return out, return_code


exec_shell("echo test running", real_time_out=True)

输出：

>>> python3 test.py
Executing 'echo test running'
test running
[OK] - The command calling was successful. CMD: 'echo test running'

希望我的回答能回答你的问题！ :)