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

在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}')

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

参考文档

后语

如果大家喜欢我写的文章,欢迎大家点赞、收藏、评论留言或者私信与我交流。

作者:namedlock原文地址:https://segmentfault.com/a/1190000043360339

%s 个评论

要回复文章请先登录注册