如何从Python异步运行外部命令？

Question

我需要从Python脚本异步运行shell命令.通过这个我的意思是我希望我的Python脚本在外部命令关闭时继续运行并执行它需要做的任何事情.

我看过这篇文章:

在Python中调用外部命令

然后我os.system()去做了一些测试,看起来我会&在命令结束时使用它来完成工作,这样我就不必等待它返回了.我想知道的是,这是否是实现这一目标的正确方法？我试过commands.call()但它对我不起作用,因为它阻止了外部命令.

如果使用os.system()这个是可取的,或者我应该尝试其他路线,请告诉我.

Answer 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上发送数据,并等待它终止.

Answer 2

如果要并行运行多个进程,然后在产生结果时处理它们,可以使用如下所示的轮询:

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第一个正在运行的进程并且结果运行时间过长,那么当您可以处理其结果时,其他正在运行的进程将处于空闲状态.

Answer 3

这在“等待命令异步终止”下的Python 3 子进程示例中涵盖：

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更多信息，请参阅。

Answer 4

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

号 os.system()是不正确的方法.这就是每个人都说使用的原因subprocess.

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

子流程模块提供了更强大的工具来生成新流程并检索其结果; 使用该模块比使用此功能更可取.使用子进程模块.请特别检查"使用子过程模块替换旧函数"部分.

Answer 5

接受的答案很旧。

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

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

并做了一些改变：

1. 让它在 Windows 上工作
2. 使其与多个命令一起工作

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 运行多个子进程并流式传输输出的方法。

Answer 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

Answer 7

Using pexpect [ http://www.noah.org/wiki/Pexpect ] with non-blocking readlines is another way to do this. Pexpect solves the deadlock problems, allows you to easily run the processes in the background, and gives easy ways to have callbacks when your process spits out predefined strings, and generally makes interacting with the process much easier.

Answer 8

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

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

但是...据我所知，这不是“完成此类事情的正确方法”，因为subprocess.CREATE_NEW_CONSOLE标志造成了安全风险。

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