如何从 Python 异步运行外部命令？-IT科技

摘要：问题描述：我需要从 Python 脚本异步运行 shell 命令。我的意思是，我希望我的 Python 脚本在外部命令执行时继续运行并执行它需要执行的任何操作。我读过这篇文章：在 Python 中调用外部命令然后我去进行了一些测试，看起来只要我在命令末尾os.system()使用，它就可以完成工作，这样我就不...

问题描述：

我需要从 Python 脚本异步运行 shell 命令。我的意思是，我希望我的 Python 脚本在外部命令执行时继续运行并执行它需要执行的任何操作。

我读过这篇文章：

在 Python 中调用外部命令

然后我去进行了一些测试，看起来只要我在命令末尾os.system()使用，它就可以完成工作，这样我就不必等待它返回。我想知道这是完成这样一件事的正确方法吗？我试过了，但它对我来说不起作用，因为它会阻止外部命令。&`commands.call()`

请让我知道是否os.system()建议使用此方法或者我是否应该尝试其他方法。

解决方案 1：

subprocess.Popen完全符合您的要求。

from subprocess import Popen
p = Popen(['watch', 'ls']) # something long running
# ... do other stuff while subprocess is running
p.terminate()

（根据评论进行编辑以完成答案）

Popen 实例可以执行各种其他操作，例如，您可以poll()查看它是否仍在运行，可以communicate()用它在 stdin 上发送数据，并等待它终止。

解决方案 2：

Python 3 子进程示例的“等待命令异步终止”下介绍了这一点。使用IPython或运行此代码python -m asyncio：

import asyncio

proc = await asyncio.create_subprocess_exec(
   'ls','-lha',
   stdout=asyncio.subprocess.PIPE,
   stderr=asyncio.subprocess.PIPE)

# do something else while ls is working

# if proc takes very long to complete, the CPUs are free to use cycles for 
# other processes
stdout, stderr = await proc.communicate()

一旦await asyncio.create_subprocess_exec(...)完成，进程就会开始运行。如果在您调用时进程尚未完成await proc.communicate()，它将等待以便为您提供输出状态。如果进程已完成，proc.communicate()将立即返回。

这里的要点与Terrels 的回答类似，但我认为 Terrels 的回答似乎使事情过于复杂。

请参阅asyncio.create_subprocess_exec以了解更多信息。

编辑：

您可以像任何其他 Python 代码一样运行上述代码，而无需-m asyncio通过使用asyncio.run()函数。

import asyncio

async def main():
    proc = await asyncio.create_subprocess_exec(
       'ls','-lha',
       stdout=asyncio.subprocess.PIPE,
       stderr=asyncio.subprocess.PIPE)

    # do something else while ls is working

    # if proc takes very long to complete, the CPUs are free to use   cycles for 
    # other processes
    stdout, stderr = await proc.communicate()
 
asyncio.run(main())

请参阅asyncio以了解更多信息

解决方案 3：

如果您想要并行运行多个进程，然后在它们产生结果时处理它们，那么您可以像下面这样使用轮询：

from subprocess import Popen, PIPE
import time

running_procs = [
    Popen(['/usr/bin/my_cmd', '-i %s' % path], stdout=PIPE, stderr=PIPE)
    for path in '/tmp/file0 /tmp/file1 /tmp/file2'.split()]

while running_procs:
    for proc in running_procs:
        retcode = proc.poll()
        if retcode is not None: # Process finished.
            running_procs.remove(proc)
            break
        else: # No process is done, wait a bit and check again.
            time.sleep(.1)
            continue

    # Here, `proc` has finished with return code `retcode`
    if retcode != 0:
        """Error handling."""
    handle_results(proc.stdout)

那里的控制流有点复杂，因为我试图把它弄小——你可以根据自己的喜好进行重构。:-)

这样做的好处是可以先处理早期完成的请求。如果您调用communicate第一个正在运行的进程，而该进程运行时间最长，那么当您本可以处理其他进程的结果时，它们将一直处于空闲状态。

解决方案 4：

我想知道的是这个 [os.system()] 是否是完成这样的事情的正确方法？

不。 os.system()不是正确的方法。这就是为什么每个人都说要使用subprocess。

有关更多信息，请阅读http://docs.python.org/library/os.html#os.system

subprocess 模块提供了更强大的功能，用于生成新进程并检索其结果；使用该模块比使用此功能更可取。请使用 subprocess 模块。请特别检查使用 subprocess 模块替换旧功能部分。

解决方案 5：

接受的答案非常古老。

我在这里找到了一个更好的现代答案：

https://kevinmccarthy.org/2016/07/25/streaming-subprocess-stdin-and-stdout-with-asyncio-in-python/

并做了一些修改：

使其在 Windows 上运行
使其与多个命令一起工作

import sys
import asyncio

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())


async def _read_stream(stream, cb):
    while True:
        line = await stream.readline()
        if line:
            cb(line)
        else:
            break


async def _stream_subprocess(cmd, stdout_cb, stderr_cb):
    try:
        process = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )

        await asyncio.wait(
            [
                _read_stream(process.stdout, stdout_cb),
                _read_stream(process.stderr, stderr_cb),
            ]
        )
        rc = await process.wait()
        return process.pid, rc
    except OSError as e:
        # the program will hang if we let any exception propagate
        return e


def execute(*aws):
    """ run the given coroutines in an asyncio loop
    returns a list containing the values returned from each coroutine.
    """
    loop = asyncio.get_event_loop()
    rc = loop.run_until_complete(asyncio.gather(*aws))
    loop.close()
    return rc


def printer(label):
    def pr(*args, **kw):
        print(label, *args, **kw)

    return pr


def name_it(start=0, template="s{}"):
    """a simple generator for task names
    """
    while True:
        yield template.format(start)
        start += 1


def runners(cmds):
    """
    cmds is a list of commands to excecute as subprocesses
    each item is a list appropriate for use by subprocess.call
    """
    next_name = name_it().__next__
    for cmd in cmds:
        name = next_name()
        out = printer(f"{name}.stdout")
        err = printer(f"{name}.stderr")
        yield _stream_subprocess(cmd, out, err)


if __name__ == "__main__":
    cmds = (
        [
            "sh",
            "-c",
            """echo "$SHELL"-stdout && sleep 1 && echo stderr 1>&2 && sleep 1 && echo done""",
        ],
        [
            "bash",
            "-c",
            "echo 'hello, Dave.' && sleep 1 && echo dave_err 1>&2 && sleep 1 && echo done",
        ],
        [sys.executable, "-c", 'print("hello from python");import sys;sys.exit(2)'],
    )

    print(execute(*runners(cmds)))

示例命令不太可能在您的系统上完美运行，并且它无法处理奇怪的错误，但此代码确实演示了一种使用 asyncio 运行多个子进程并流式传输输出的方法。

解决方案 6：

我使用asyncproc模块取得了很大成功，它可以很好地处理进程的输出。例如：

import os
from asynproc import Process
myProc = Process("myprogram.app")

while True:
    # check to see if process has ended
    poll = myProc.wait(os.WNOHANG)
    if poll is not None:
        break
    # print any new output
    out = myProc.read()
    if out != "":
        print out

解决方案 7：

使用pexpect和非阻塞 readlines 是实现此目的的另一种方法。Pexpect 解决了死锁问题，允许您轻松地在后台运行进程，并在您的进程吐出预定义字符串时提供简单的回调方法，并且通常使与进程的交互变得更加容易。

解决方案 8：

考虑到“我不必等待它返回”，最简单的解决方案之一是：

subprocess.Popen( \n    [path_to_executable, arg1, arg2, ... argN],
    creationflags = subprocess.CREATE_NEW_CONSOLE,
).pid

但是...从我所读的内容来看，这不是“完成这样的事情的正确方法”，因为subprocess.CREATE_NEW_CONSOLE标志会产生安全风险。

这里发生的关键事情是使用subprocess.CREATE_NEW_CONSOLE来创建新的控制台和.pid（返回进程 ID，以便您可以稍后检查程序），这样就不必等待程序完成其工作。

解决方案 9：

我在尝试使用 Python 中的 s3270 脚本软件连接到 3270 终端时遇到了同样的问题。现在，我正在使用我在这里找到的 Process 子类来解决问题：

http://code.activestate.com/recipes/440554/

以下是从文件中获取的样本：

def recv_some(p, t=.1, e=1, tr=5, stderr=0):
    if tr < 1:
        tr = 1
    x = time.time()+t
    y = []
    r = ''
    pr = p.recv
    if stderr:
        pr = p.recv_err
    while time.time() < x or r:
        r = pr()
        if r is None:
            if e:
                raise Exception(message)
            else:
                break
        elif r:
            y.append(r)
        else:
            time.sleep(max((x-time.time())/tr, 0))
    return ''.join(y)

def send_all(p, data):
    while len(data):
        sent = p.send(data)
        if sent is None:
            raise Exception(message)
        data = buffer(data, sent)

if __name__ == '__main__':
    if sys.platform == 'win32':
        shell, commands, tail = ('cmd', ('dir /w', 'echo HELLO WORLD'), '
')
    else:
        shell, commands, tail = ('sh', ('ls', 'echo HELLO WORLD'), '
')

    a = Popen(shell, stdin=PIPE, stdout=PIPE)
    print recv_some(a),
    for cmd in commands:
        send_all(a, cmd + tail)
        print recv_some(a),
    send_all(a, 'exit' + tail)
    print recv_some(a, e=0)
    a.wait()

解决方案 10：

这里有几个答案，但没有一个满足我的以下要求：

我不想等待命令完成或用子进程输出污染我的终端。
我想运行带有重定向的 bash 脚本。
我想在我的 bash 脚本中支持管道（例如find ... | tar ...）。

唯一满足上述要求的组合是：

subprocess.Popen(['./my_script.sh "arg1" > "redirect/path/to"'],
                 stdout=subprocess.PIPE, 
                 stderr=subprocess.PIPE,
                 shell=True)