如何在subprocess Popen的communicate执行期间持续打印‘Processing...’

你的代码核心问题在于：communicate()是阻塞调用，会一直等待进程结束才返回，而你之前的while p.poll()循环会在调用communicate()前执行，两者无法同时运行。要实现communicate()执行期间持续打印提示，最直接的方式是用线程分离打印逻辑和进程输入输出处理逻辑。

另外，你提到命令执行结束后需要按下回车键才能关闭，因此在communicate()时需要给进程的标准输入写入换行符。

完整解决方案代码 #

import subprocess
import threading
import time

def print_processing(process):
    # 持续打印直到进程结束
    while process.poll() is None:
        # 用\r覆盖当前行，避免刷屏
        print("Processing...", end='\r', flush=True)
        time.sleep(1)  # 控制打印频率，避免资源浪费

# 替换为你的实际命令列表
abc = ["your_command_here"]
# 启动子进程
p = subprocess.Popen(
    abc,
    stdout=subprocess.PIPE,
    stdin=subprocess.PIPE,
    stderr=subprocess.PIPE,
    shell=False
)

# 启动打印线程
processing_thread = threading.Thread(target=print_processing, args=(p,))
processing_thread.start()

# 给进程发送回车键，满足命令结束后的输入要求
# 注意：如果命令需要的是其他输入，替换成对应字节内容
output, error = p.communicate(input=b'\n')

# 等待打印线程结束
processing_thread.join()

# 处理并打印结果
print("\n执行完成")
print("标准输出:\n", output.decode('utf-8'))
print("错误输出:\n", error.decode('utf-8'))

关键说明 #

1. 线程分离逻辑：打印逻辑放在单独线程中，主线程专注于communicate()处理进程的输入输出，两者可以同时运行。
2. 避免刷屏：使用end='\r'让光标回到行首，每次打印的“Processing...”会覆盖之前的内容，同时加上flush=True确保内容立即输出到控制台。
3. 处理回车键需求：communicate(input=b'\n')给子进程的标准输入写入换行符，模拟按下回车键的操作，确保进程能正常结束。
4. 控制打印频率：time.sleep(1)让每秒打印一次，避免频繁打印占用过多资源。

内容的提问来源于stack exchange，提问作者Mahnaz Rafia Islam