在线程上下文中捕获stdout/stderr和函数返回值

Question

我需要能够在线程上下文中捕获函数的返回值及其stdout/stderr。目前，我正在通过调用每个子类中的is_true方法来评估基于这个基类（my_base_class）的许多子类（my_class_foo，my_class_bar）。你知道吗

我希望能够捕获每个is_true的返回值以及stdout/stderr。下面的当前解决方案在非线程或非多进程上下文中工作。但是，它依赖于重定向stdout/stderr，如果我并行计算多个子类，显然这是行不通的。你知道吗

我已经查看了concurrent.futures和multiprocess和subprocess包，无法找到解决方案。你知道吗

我试图避免使用记录器，这样用户就可以只依赖于打印到stdout，而不是使用显式方法。你知道吗

我想并行地执行来自my_class_foo和my_class_bar的is_true方法，并且能够用每个类的返回值捕获stdout消息。你知道吗

class my_class_foo(my_base_class):
    def is_true(self):
        print('foo')
        return True


class my_class_bar(my_base_class):
    def is_true(self):
        print('bar')
        return False


class my_base_class(object):

    def is_true(self):
        Raise NotImplementedError

    def evaluate_node_is_true(self):
        with Capturing() as is_true_stdout:
            node_is_true = self.is_true()
            self.output = ''.join(is_true_stdout)


class Capturing(list):
    """
    Context manager for capturing the stdout of the is_true() function call
    """
    def __enter__(self):
        self._stdout = sys.stdout
        self._stderr = sys.stderr
        sys.stdout = self._stringio_out = io.StringIO()
        sys.stderr = self._stringio_err = io.StringIO()
        return self
    def __exit__(self, *args):
        self.extend(self._stringio_out.getvalue().splitlines())
        self.extend(self._stringio_err.getvalue().splitlines())
        del self._stringio_out
        del self._stringio_err
        sys.stdout = self._stdout
        sys.stderr = self._stderr

Answer 1

最简单的解决方案可能是subprocess。当然，如果你想和孩子们共享任何数据，这是行不通的。它需要为子进程编写一个简单的独立驱动程序脚本，而不是依赖与主程序相同的脚本。但如果成功了，就这么简单：

res = subprocess.run([sys.executable, driver_script_name],
                     stdout=subprocess.PIPE, stderr=subprocess.PIPE)

然后你就有res.stdout和res.stderr可以阅读了。你知道吗

要同时执行多个子线程，最简单的解决方案是为每个subprocess.run触发一个线程。如果您想要一个一次只有8个的池，那么使用ThreadPoolExecutor。你知道吗

对于像您这样的问题，subprocess的最大问题是需要stdout、stderr和返回值。进程会给您stdout、stderr和一个返回码，但该码只是一个8位数字。你知道吗

在您的例子中（基于您的注释），返回值实际上是一个bool，这样就可以了。Unix命令行工具有一个准标准，它执行布尔操作，返回0表示true，返回1表示false，就像使用true和false工具一样。你知道吗

这与您可能预期的相反，但它符合更一般的准标准，即0表示成功，1表示一般错误，2表示参数错误，3-127表示工具特定错误，128-255不使用，因为如果被信号杀死，有时会丢失最后一位。如果您（或您的用户）可能希望从shell测试子程序，请使用0表示true。你知道吗

---

你也可以用multiprocessing来实现，但这有点棘手。您可以手动创建管道并通过stdio复制它们，但这很难正确实现。更糟糕的是，我不认为这种方法是有文档记录的，所以您必须深入到源代码中multiprocessing才能做到这一点。你知道吗

---

事实上，至少如果您使用*nix，在这种情况下手动fork可能会更容易。当然，它是低级的，而且有很大的出错空间，但是（假设您对Unix fork非常了解）您确切地知道要做什么和在哪里做。你知道吗

---

对于multiprocessing来说，一个更简单的选择是执行this answer演示的操作，让每个子级将其stdout重定向到一个文件，然后父级可以稍后读取该文件。你知道吗

---

一个甚至适用于线程的选项是用附加到线程特定缓冲区的自定义类文件对象替换sys.stdout。您可以使用线程本地数据，然后在最后将其复制回来，但只使用由线程id键入的dict可能更简单

class ThreadedStdWriter(io.RawIOBase):
    ThreadedStdWriter.buffers = {}
    def __init__(self):
        ThreadedStdWriter.buffers[threading.get_ident()] = []
    def write(self, b):
        ThreadedStdWriter.buffers.append(b)

…然后将其包装成BufferedWriter和TextIOWrapper并将结果存储为sys.stdout，然后对sys.stderr执行相同的操作。那么，任何线程的stdout就是b''.join(ThreadedStdWriter.buffers[thread_id])。你知道吗