Python3多线程场景下避免CLI命令输入被终端输出打断的方案问询

问题描述 #

我正在开发一款客户端/服务端CLI应用，遇到Python3的input()函数与多线程的冲突问题。示例代码如下：

import threading
import time

def thread_function():
    for i in range(10):
        time.sleep(3)
        print(f"[Thread {i}]")


while True:
    thread = threading.Thread(target=thread_function)
    thread.start()
    command = input("input> ")
    print("Command:",super1)

问题现象 #

该代码在提示用户输入命令并输出的同时，有一个线程会定时（每3秒）在终端输出内容。当输入命令时，线程输出会打断输入，导致输入内容被拆分，例如输入whoami时会出现：

input> who[Thread 0]
ami
Command:whoami

期望效果 #

输入到一半时线程输出内容后，能继续在新的输入提示符后完成命令输入，且需支持退格键。同时线程输出必须实时显示，不能等待输入结束。期望效果如下：

input> who
[Thread 0]
input> whoami

尝试过的方案 #

曾尝试使用sys.stdin和readchar，但未找到合适的实现方式：

• 使用readchar时难以同步字符列表与线程输出，且无法处理退格键
• 使用sys.stdin则无法逐字符获取完整命令

补充背景：项目为LightC2，问题出现在libs.client.cliclient脚本的interact_with_agent函数的while循环中，需要随时获取命令输出，同时不被代理输出打断输入。

最终可行解决方案（未优化） #

以下是找到的可行解决方案代码：

import readchar
from threading import Thread
import threading
import time

global command
command = ""
lock = threading.Lock()

def print_various(printlock):
    global command
    for _ in range(10):
        time.sleep(2)
        with printlock:
            print("\n"+"[Ok From Thread]"+f"\nInput >{command}",end="")

thread1 = Thread(target=print_various,args=[lock])
thread1.start()

while True:
    with lock:
        print("Input >",end="",flush=True)
    command = ""
    while True:
        try:
            char = readchar.readkey()
        except KeyboardInterrupt:
            command = ""

        if char=="\n":
            print("\n"+command,flush=True)
            break

        if command=="exit":
            break

        elif char==readchar.key.BACKSPACE:
            command = command[:-1]
            with lock:
                print("\r",end='\x1b[2K')
                print(f"Input >{command}",end="",flush=True)
            
        else:
            command+=char
            with lock:
                print(char,end="",flush=True)
    if command == "exit":
        break

完整的类实现可查看LightC2项目仓库。

内容的提问来源于stack exchange，提问作者Ph3nX