Windows 上的异步子进程

首先，我要解决的总体问题比我在这里展示的要复杂一些，所以请不要告诉我“使用阻塞线程”，因为如果没有公平、公平的重写，它就无法解决我的实际情况重构。

我有几个不需要我修改的应用程序，它们从标准输入获取数据，并在发挥其魔力后将其输出到标准输出。我的任务是链接其中几个程序。问题是，有时他们会窒息，因此我需要跟踪他们在 STDERR 上输出的进度。

pA = subprocess.Popen(CommandA,  shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
# ... some more processes make up the chain, but that is irrelevant to the problem
pB = subprocess.Popen(CommandB, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=pA.stdout )

现在，直接通过 pA.stdout.readline() 和 pB.stdout.readline() 或普通的 read() 函数读取是一个阻塞问题。由于不同的应用程序以不同的速度和不同的格式输出，因此不能选择阻塞。 （正如我上面所写的，除非万不得已，否则线程不是一个选择。）pA.communicate()是死锁安全的，但由于我需要实时信息，所以这也不是一个选择。

因此谷歌把我带到了这个异步子流程片段 http://code.activestate.com/recipes/440554-module-to-allow-asynchronous-subprocess-use-on-win/在活动状态上。

一开始一切都很好，直到我实施它。比较 cmd.exe 的输出pA.exe | pB.exe，忽略两个输出到同一窗口造成混乱的事实，我看到非常即时的更新。但是，我使用上面的代码片段和read_some()函数在那里声明，并且需要 10 秒以上才能通知单个管道的更新。但当它发生时，它会进行更新，例如将进度一路提升至 40%。

因此，我做了一些更多的研究，并看到许多有关 PeekNamedPipe、匿名句柄以及即使管道中有可用信息也返回 0 字节的主题。由于事实证明这个主题远远超出了我的专业知识，无法修复或编码，因此我来到 Stack Overflow 寻求指导。 :)

我的平台是 W7 64 位和 Python 2.6，应用程序是 32 位（以防万一），并且与 Unix 的兼容性不是问题。我什至可以处理一个完整的 ctypes 或 pywin32 解决方案，如果它是唯一的解决方案，它可以完全颠覆子进程，只要我可以异步地从每个 stderr 管道中读取数据，并立即执行性能并且没有死锁。 :)

必须使用线程有多糟糕？我遇到了很多同样的问题，最终决定使用线程来收集子进程的 stdout 和 stderr 上的所有数据，并将其放入线程安全队列中，主线程可以以阻塞方式读取该队列，而不必担心幕后发生的线程。

目前尚不清楚基于线程和阻塞的解决方案预计会出现什么问题。您是否担心必须使其余代码成为线程安全的？这不应该成为问题，因为 IO 线程不需要与任何其余代码或数据交互。如果您的内存要求非常严格，或者您的管道特别长，那么您可能会对生成如此多的线程感到不高兴。我对你的情况不太了解，所以我不能说这是否可能是一个问题，但在我看来，既然你已经产生了额外的进程，那么与它们交互的一些线程不应该是一个可怕的负担。在我的情况下，我还没有发现这些 IO 线程特别有问题。

我的线程函数看起来像这样：

def simple_io_thread(pipe, queue, tag, stop_event):
    """
    Read line-by-line from pipe, writing (tag, line) to the
    queue. Also checks for a stop_event to give up before
    the end of the stream.
    """
    while True:
        line = pipe.readline()

        while True:
            try:
                # Post to the queue with a large timeout in case the
                # queue is full.
                queue.put((tag, line), block=True, timeout=60)
                break
            except Queue.Full:
                if stop_event.isSet():
                    break
                continue
        if stop_event.isSet() or line=="":
            break
    pipe.close()

当我启动子进程时，我这样做：

outputqueue = Queue.Queue(50)
stop_event = threading.Event()
process = subprocess.Popen(
    command,
    cwd=workingdir,
    env=env,
    shell=useshell,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE)
stderr_thread = threading.Thread(
    target=simple_io_thread,
    args=(process.stderr, outputqueue, "STDERR", stop_event)
)
stdout_thread = threading.Thread(
    target=simple_io_thread,
    args=(process.stdout, outputqueue, "STDOUT", stop_event)
)
stderr_thread.daemon = True
stdout_thread.daemon = True
stderr_thread.start()
stdout_thread.start()

然后，当我想读取时，我可以阻塞输出队列 - 从它读取的每个项目都包含一个字符串来标识它来自哪个管道，以及来自该管道的一行文本。很少有代码在单独的线程中运行，并且它仅通过线程安全队列与主线程通信（加上一个事件，以防我需要提前放弃）。也许这种方法很有用，可以让您通过线程和阻塞来解决问题，但不必重写大量代码？

（我的解决方案变得更加复杂，因为我有时希望尽早终止子进程，并希望确保线程全部完成。如果这不是问题，您可以摆脱所有 stop_event 内容，它会变得非常简洁。）