在 Python 中使用异步方式执行 `subprocess` 来运行 `.bat` 文件可以通过 `asyncio` 模块与 `subprocess` 模块的结合实现。`asyncio` 提供了 `asyncio.create_subprocess_exec` 或 `asyncio.create_subprocess_shell` 方法来创建子进程，并通过 `await` 关键字实现异步等待其完成。 以下是一个使用 `asyncio` 和 `subprocess` 异步运行 `.bat` 文件的示例代码： ```python import asyncio async def run_bat_file(): # 假设你的 .bat 文件路径为 'example.bat' process = await asyncio.create_subprocess_shell( 'example.bat', stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) # 等待进程完成并获取输出 stdout, stderr = await process.communicate() # 打印输出 if stdout: print(f'[stdout]\n{stdout.decode()}') if stderr: print(f'[stderr]\n{stderr.decode()}') # 运行异步任务 asyncio.run(run_bat_file()) ``` ### 说明： - `asyncio.create_subprocess_shell()` 用于通过系统 shell 执行命令，适用于包含 shell 特性的命令。如果不需要 shell 特性，可以使用 `asyncio.create_subprocess_exec()`，它直接执行可执行文件。 - `stdout=asyncio.subprocess.PIPE` 和 `stderr=asyncio.subprocess.PIPE` 用于捕获标准输出和标准错误。 - `process.communicate()` 用于读取子进程的标准输出和标准错误，并等待其完成。 - `asyncio.run()` 是运行异步函数的标准方法，适用于 Python 3.7 及以上版本。 ### 超时处理 如果希望为子进程设置超时时间，可以在 `await process.communicate()` 时使用 `asyncio.wait_for()` 包裹并设置 `timeout` 参数： ```python try: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=10) except asyncio.TimeoutError: print("子进程执行超时，正在终止...") process.kill() await process.wait() ``` ### 性能优化 对于需要频繁运行外部命令的应用场景，可以考虑将异步子进程调用封装成一个可复用的函数，例如： ```python async def async_run_command(command: str, timeout: int = 10): process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) try: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout) return stdout.decode(), stderr.decode(), process.returncode except asyncio.TimeoutError: print(f"命令 {command} 超时，正在终止...") process.kill() await process.wait() return None, "Timeout", -1 ``` ### 使用示例 ```python async def main(): stdout, stderr, returncode = await async_run_command('example.bat') print(f"标准输出: {stdout}") print(f"标准错误: {stderr}") print(f"返回码: {returncode}") asyncio.run(main()) ``` 这种方式可以将子进程的执行、超时处理和输出捕获封装为通用函数，便于在多个异步任务中复用。