### 解决 `subprocess.run` 中 `input()` 阻塞问题 当使用 `subprocess.run` 来执行包含 `input()` 的 Python 脚本时，可能会遇到输入阻塞的情况。这是因为默认情况下，子进程的标准输入被重定向到了父进程中，而没有提供任何输入数据给子进程。 为了处理这种情况，可以通过设置 `stdin` 参数来传递所需的输入数据。下面是一个完整的解决方案： #### 方法一：通过字符串传递输入 如果已知需要向脚本发送的具体输入内容，则可以直接将其作为字节串传入 `communicate()` 或者直接在调用 `run` 时指定 `input` 参数[^1]。 ```python import subprocess result = subprocess.run( ['python3', 'script_with_input.py'], input=b'yes\n', stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True ) print(result.stdout.decode()) ``` 此方法适用于简单的命令行交互场景，在这种情况下能够有效地避免因等待用户输入而导致的程序挂起现象。 #### 方法二：实时读写标准流 对于更复杂的交互需求，比如连续多次提示用户输入或者根据不同的输出做出反应，应该考虑采用异步 I/O 技术或利用多线程/多进程模型来进行并发操作[^3]。 一种常见做法是创建一个新的线程专门负责监听并响应来自子进程的消息；另一个则是借助于像 asyncio 这样的库实现事件驱动型编程模式下的非阻塞通信机制。 这里给出一个基于线程池的例子： ```python from concurrent.futures import ThreadPoolExecutor import sys import threading import queue import subprocess def enqueue_output(out, que): for line in iter(out.readline, b''): que.put(line) out.close() cmd = ["python3", "-c", "for i in range(5): print(input(f'Type something {i}:'))"] proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, bufsize=0) q_stdout = queue.Queue() t_stdout = threading.Thread(target=enqueue_output, args=(proc.stdout, q_stdout)) t_stdout.daemon = True t_stdout.start() with ThreadPoolExecutor(max_workers=2) as executor: future_to_idx = { executor.submit(proc.stdin.write, f"{idx}\n".encode()): idx for idx in range(5) } while not all([f.done() for f in future_to_idx]): try: line = q_stdout.get_nowait().decode().strip() print(line) except queue.Empty: pass proc.stdin.flush() proc.terminate() ``` 上述代码展示了如何在一个独立的工作线程中持续不断地从子进程的标准输出管道获取最新消息，并及时反馈至主线程以便进一步处理。与此同时，还安排了另外一组工作单元用于周期性地向目标应用程序注入必要的控制指令序列。