1.问题
并发性能一直都是Python之殇
,线程很多情况下不能提升性能,而且不容易杀死,容易阻塞,编写线程并发永远不是一个简单的问题。多进程并发,也是如此,如何杀死进程,如何读进程输出不阻塞,处理起来也不是那么容易。本文一起看看读进程输出时,如何防止阻塞,不导致父进程挂死的处理方法。
2.方案
2.1.创建子进程
Python
中创建子进程,还是非常容易的。我们可以创建dir进程。代码如下:
import subprocessimport timeimport shlexfrom loguru import loggercmd = 'dir'args = shlex.split(cmd)proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, shell=True, universal_newlines=False)data = proc.stdout.read()logger.info(data.decode('cp936'))
上述代码可能的输出如下:
PS C:\code\python\pylearn> python3 main.pyPS C:\code\python\pylearn> PS C:\code\python\pylearn>PS C:\code\python\pylearn> dir 目录: C:\code\python\pylearn C:\code\python\pylearn 的目录2023/03/29 23:18 <DIR> .2023/03/29 23:12 <DIR> ..2023/03/29 23:16 <DIR> .idea2023/03/29 23:18 359 main.py 1 个文件 359 字节 3 个目录 226,689,306,624 可用字节PS C:\code\python\pylearn>
- 可通过
subprocess.Popen
创建子进程,同时使用subprocess.PIPE
把子进程的输入输出重新导向到管道中,这样我们就可以通过管道和子进程进行交互了。 - 使用
proc.stdout.read()
获取子进程的输出。
上述是读进程的输出,看上去没什么问题,但是如果换一个进程cmd.exe
,这个时候程序就挂死了,原因是管道读proc.stdout.read()
是阻塞式的,只有子进程退出时才会返回结果,很容易挂死。虽然官方也有API进行枚举子进程是否有输出(比如:communicate
,有兴趣去可以搜索下,不是本文的重点),使用起来限制太多,不容易设计成与子进程进行灵活的交互。
2.2.子进程无阻塞读(windows
版本)
Python
中进行无阻塞式读,没有标准的API可以用。不过好在,主流的Python解释器
是C语言开发的,我们可以直接访问windows
系统的API把管道设置成无阻塞。
使用windows
的API有两种,一种使用第三方库pywin32
;另外一种使用Python
自带的内部API。我们先来看代码:
import _winapiimport msvcrthandle = msvcrt.get_osfhandle(proc.stdout.fileno())read, avail_count, msg = _winapi.PeekNamedPipe(handle, 0)if avail_count > 0: data, errcode = _winapi.ReadFile(handle, avail_count) logger.info(data.decode('cp936'))else: return b''
msvcrt.get_osfhandle
方法,是把子进程的管道转换成windows
系统的句柄。_winapi.PeekNamedPipe
方法,可以查看下句柄是否有数据可读取,如果有数据可读的话,返回值avail_count
会大于0(备注:PeekNamedPipe
返回的是元组)。_winapi.ReadFile
方法,就是windows
系统标准的读数据的方法,有点类似linux
系统中的read
方法。
windows
系统API,在Python
中使用起来是比较简单的,远比C语言中使用容易。但是问题又来了,linux
中我们又如何处理呢?
2.3.子进程无阻塞读(linux版本)
好消息是,linux
系统中想要对管道进行非阻塞式读是很容易的,有标准的库可以使用,我们直接上代码,如下:
import fcntlflags = fcntl.fcntl(proc.stdout, fcntl.F_GETFL)fcntl.fcntl(proc.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK)flags = fcntl.fcntl(proc.stderr, fcntl.F_GETFL)fcntl.fcntl(proc.stderr, fcntl.F_SETFL, flags | os.O_NONBLOCK)
fcntl.fcntl
方法,很容易把管道设置成非阻塞方式,该方法的详细使用说明,大家搜索查找帮助,这里不详细说明如何使用。
2.4.子进程写
子进程的输入管道,写一般不会阻塞的,直接调用write
方法,无非就是str和bytes
类型的转换而已。代码如下:
proc.stdin.write(b'dir\r\n')proc.stdin.flush()
- 唯一要注意的时候,调用
wirte
方法,需要加上回车换行\r\n
,才会真正写入到子进程。
3.讨论
到这里为止,关于子进程读写的主要内容都已经说到了,剩下的就是进行详细的封装,设计成类,便于使用而已。详细的代码,这里我也放上来,大家有兴趣学习研究。
class ChildProcess: def __init__(self): self._child_process = None def __del__(self): if not self._child_process: return self.kill_process() def create_process(self, cmd: str, shell=False): args = shlex.split(cmd) self._child_process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, shell=shell, universal_newlines=False) result = self._child_process.stdout.readable() if result: print('can read') if not OsUtils.is_windows(): # 设置非阻塞 flags = fcntl.fcntl(self._child_process.stdout, fcntl.F_GETFL) fcntl.fcntl(self._child_process.stdout, fcntl.F_SETFL, flags | os.O_NONBLOCK) flags = fcntl.fcntl(self._child_process.stderr, fcntl.F_GETFL) fcntl.fcntl(self._child_process.stderr, fcntl.F_SETFL, flags | os.O_NONBLOCK) else: pass if self._child_process.poll() is not None: logger.warning(f'create process: {cmd} ok, process id: {self._child_process.pid} has exist') else: logger.info(f"create process: {cmd} ok, process id: {self._child_process.pid}") return False def is_process_exit(self): if self._child_process.poll() is not None: return False else: return True def kill_process(self): if not self._child_process: return if not self.is_process_exit(): self._child_process.wait() self._child_process.kill() self._child_process = None def write(self, cmd: str): """ 写入字节 :param cmd: :return: """ return self.write_bytes(cmd.encode('utf-8')) def write_bytes(self, data: bytes): if not self.is_process_exit(): return self._child_process.stdin.write(data) self._child_process.stdin.flush() def read(self, timeout, wait_process_exist=False): data = self.read_bytes(timeout=timeout, wait_process_exist=wait_process_exist) if OsUtils.is_windows(): return StrUtils.remove_ansi_escape(data.decode('cp936')) else: return StrUtils.remove_ansi_escape(data.decode('utf-8')) def read_bytes(self, timeout, wait_process_exist=False): @loop_exec_timeout(timeout=timeout, interval=20 / 1000, expect_return=True) def _read(): if not OsUtils.is_windows(): data = self._child_process.stdout.read() else: data = self._read_stdout_win() if not data: return False nonlocal total_recv_bytes total_recv_bytes.write(data) if wait_process_exist and self.is_process_exit(): return False return True total_recv_bytes = BytesIO() _read() return total_recv_bytes.getvalue() def read_until(self, prompt: str, prompt_re, timeout=5): @loop_exec_timeout(timeout=timeout, interval=0.1, expect_return=True) def _read_until(): nonlocal total_data data = self.read_bytes(1) if not data: return False total_data.write(data) s = total_data.getvalue().decode('utf-8') s = StrUtils.remove_ansi_escape(s) last_line = s.splitlines()[-1].rstrip() if prompt and last_line.endswith(prompt): return True if prompt_re and StrUtils.endswith_re(last_line, prompt_re): return True return False total_data = BytesIO() _read_until() return StrUtils.remove_ansi_escape(total_data.getvalue().decode('utf-8')) def _read_stdout_win(self): import _winapi import msvcrt handle = msvcrt.get_osfhandle(self._child_process.stdout.fileno()) read, avail_count, msg = _winapi.PeekNamedPipe(handle, 0) if avail_count > 0: data, errcode = _winapi.ReadFile(handle, avail_count) return data else: return b''
上述代码中loop_exec_timeout
(循环执行的装饰器)等方法都比较简单,限于篇幅就不放代码了,自己改造下,或者自己实现。
基于上述代码,可以很容易封装交互式的子进程类型,比如cmd.exe,plink.exe
等。再也不用担心子进程会该死的问题了。
如果能用标准库实现,我尽量不会用第三库,比如本文中,一般会想到使用pywin32
库调用windows
系统的API,但这个库规模也挺大,还要安装。既然Python
本身也提供调用windows系统
的API,就不一定非得依赖pywin32
了。
Enjoy code~~~