从后台运行的 shell 命令获取输出

Question

我在this post 中看到了一些有用的信息，说明如果您使用subprocess 从中检索输出，您不能期望在后台运行进程。问题是……这正是我想要做的！

我有一个脚本，它通过ssh 将命令发送到各个主机，我不想在开始下一个之前等待每个主机完成。理想情况下，我可以有这样的东西：

for host in hostnames:
  p[host] = Popen(["ssh", mycommand], stdout=PIPE, stderr=PIPE)
  pout[host], perr[host] = p[host].communicate()

这将有（在mycommand 需要很长时间的情况下）所有主机同时运行mycommand。就像现在一样，整个 ssh 命令似乎在开始下一个命令之前完成。这是（根据我链接的上一篇文章）由于我正在捕获输出，对吗？除了cat将输出写入文件并稍后读取输出之外，有没有一种体面的方法可以让这些事情在不同的主机上并行发生？

Answer 1

您可能想为此使用fabric。

Fabric 是一个 Python (2.5-2.7) 库和命令行工具，用于简化 SSH 在应用程序部署或系统管理任务中的使用。

示例文件：

from fabric.api import run, env

def do_mycommand():
    my_command = "ls" # change to your command
    output = run(mycommand)
    print "Output of %s on %s:%s" % (mycommand, env.host_string, output)

现在在所有主机上执行（host1,host2 ... 是所有主机所在的位置）：

fab -H host1,host2 ... do_mycommand

Answer 2

您可以使用threads 实现并行性，使用Queue 以线程安全的方式检索结果：

import subprocess
import threading
import Queue

def run_remote_async(host, command, result_queue, identifier=None):
    if isinstance(command, str):
        command = [command]

    if identifier is None:
        identifier = "{}: '{}'".format(host, ' '.join(command))

    def worker(worker_command_list, worker_identifier):
        p = subprocess.Popen(worker_command_list,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE)
        result_queue.put((worker_identifier, ) + p.communicate())

    t = threading.Thread(target=worker,
            args=(['ssh', host] + command, identifier),
            name=identifier)
    t.daemon = True
    t.start()

    return t

那么，一个可能的测试用例可能如下所示：

def test():
    data = [('host1', ['ls', '-la']),
            ('host2', 'whoami'),
            ('host3', ['echo', '"Foobar"'])]
    q = Queue.Queue()
    for host, command in data:
        run_remote_async(host, command, q)
    for i in range(len(data)):
        identifier, stdout, stderr = q.get()
        print identifier
        print stdout

Queue.get() 处于阻塞状态，因此此时您可以在任务完成后收集一个接一个的结果。