Python 如何创建一个超时后不会被 kill 的 python 子进程

namedlock · January 29, 2023 · Last by namedlock replied at January 30, 2023 · 4754 hits

 在 subprocess 之前,创建一个新进程的方式有很多种,如 os.system()、os.spawn* 方法等。为了统一创建进程的方式,python 社区提议使用 subprocess 模块来统一创建进程,替换 os.system 和 os.spawn*。

subprocess 模块的使用

subprocess.run

方法签名如下

subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, capture_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal_newlines=None, **other_popen_kwargs)

run 方法是创建进程的推荐方法,只有 run 方法无法满足需要的时候才考虑使用 popen 方法。run 会堵塞式地执行 args 参数提供的命令,直到命令执行结束或者超时。args 可以是字符串数组或者字符串,当 args 为字符串时,shell 参数必须为 True

 

我想创建一个子进程,使用 adb install 命令给 vivo/oppo/xiaomi 手机安装 app,但是 vivo/oppo/xiaomi 手机会弹出二次确认窗口,如果不进行点击操作,则无法安装 App。所以我就想创建一个子进程,超时后再接着由 UI 自动化的方式来点击确认按钮。这就要求超时后安装进程不能退出。run 方法虽然有 timeout 参数,但是子进程会在超时被 kill 掉。

以下动画演示了手动安装 app 时需要二次确认,手动点击继续安装后,可以在控制台看到 Perform Streamed Install Success 的输出。

 

以下动画演示了使用 run 命令,超时后安装进程被 kill 掉了,在控制台无法看到成功安装的日志输出。

对应的代码:

import subprocess

if __name__ == '__main__':

    proc=subprocess.run('adb install -g -r -t app-uiautomator.apk', shell=True, timeout=10, capture_output=True)

 

subprocess.Popen

subprocess 模块里进程的创建和管理都是 Popen 类处理的,它让开发者非常灵活地处理一些不常见的场景。

 

class subprocess.Popen(args, bufsize=- 1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, group=None, extra_groups=None, user=None, umask=- 1, encoding=None, errors=None, text=None, pipesize=- 1, process_group=None)

Popen 类有以下几个方法:

  • Popen.communicate

 

Popen.communicate(input=None, timeout=None)
communicate()返回一个元组(stdout_data, stderr_data),如果Popen指定了text模式,stdout_data将为字符串,否则为byte
  1. input 向标准输入发送信息

  2. timeout 如果进行在 timeout 指定的时间之内没有结束,则抛出一个 TimeoutExpired 异常,且进程不会被杀死

抛出异常后可以继续重新调用 communicate()

 

* Popen.send_signal(signal) 向子进程发送信号

* Popen.terminate() 停止子进程

* Popen.poll() 检查进程是否结束,没有结束返回 None,否则返回执行状态的数值

* Popen.wait(timeout=None) 等待子进程执行结束,如果 timeout 指定的时间之后进程没有结束,则抛出 TimeoutExpired

 

 在我的执行场景中,需要使用 communicate 方法,以下是使用 Popen 类调用 adb 命令安装 app 的动画展示,超时后,手动点击继续安装,app 最终安装成功。

import subprocess

from subprocess import TimeoutExpired

 

if __name__ == '__main__':

    try:

        proc=subprocess.Popen('adb install -g -r -t app-uiautomator.apk', shell=True, text=True, stdout=subprocess.PIPE)

        stdout,errs = proc.communicate(timeout=10)

        print(f'stdout1:{stdout}')

    except TimeoutExpired as te:

        print('timeout')

        stdout,errs = proc.communicate()

        print(f'stdout2:\n{stdout}')

 

这样的话,我的目的就达到了。

参考文档

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
最佳回复

再写个进程监控手机屏幕嘛, 发现弹窗就自动点击.

共收到 2 条回复 时间 点赞

再写个进程监控手机屏幕嘛, 发现弹窗就自动点击.

fishzhang 回复

😀 可以的

namedlock vscode 中调试 python 代码 中提及了此贴 30 Jan 21:32
需要 Sign In 后方可回复, 如果你还没有账号请点击这里 Sign Up