1.问题
并发性能一直都是Python之殇，线程很多情况下不能提升性能，而且不容易杀死，容易阻塞，编写线程并发永远不是一个简单的问题。多进程并发，也是如此，如何杀死进程，如何读进程输出不阻塞，处理起来也不是那么容易。本文一起看看读进程输出时，如何防止阻塞，不导致父进程挂死的处理方法。
2.方案
2.1.创建子进程
Python中创建子进程，还是非常容易的。我们可以创建dir进程。代码如下：
import subprocess
import time
import shlex
from loguru import logger
cmd = '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.py
PS 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> .idea
2023/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 _winapi
import msvcrt
handle = 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 fcntl
flags = 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~~~
