Python多线程print输出与控制台input输入重叠的跨平台解决方案咨询

你遇到的问题本质是input和后台print没有同步机制，二者同时向标准输出写入内容时会打断输入行渲染，导致内容重叠。以下提供两种适配Windows优先、兼容Linux的解决方案：

方案一：使用成熟第三方库prompt_toolkit（推荐生产环境使用） #

该方案无需自行处理底层终端控制逻辑，prompt_toolkit已原生实现跨平台终端适配，自动处理后台打印与输入行的展示优先级，完美匹配你需要的输出效果。

安装依赖 #

pip install prompt_toolkit

完整实现代码 #

import threading
import random
import time
from prompt_toolkit import PromptSession
from prompt_toolkit import print_formatted_text as print

def some_process():
    for i in range(50):
        time.sleep(random.randint(1, 3))
        print("Some program output")

if __name__ == "__main__":
    session = PromptSession()
    thread = threading.Thread(target=some_process, daemon=True)
    thread.start()
    
    while True:
        try:
            shell = session.prompt("> ")
            # 此处处理你的指令逻辑，如help、exit等
            if shell.strip() == "exit":
                break
            print(f"执行指令: {shell}")
        except KeyboardInterrupt:
            # 处理Ctrl+C主动退出逻辑
            break

方案优势 #

• 完全跨平台，适配Windows下的cmd、PowerShell、Windows Terminal，以及Linux所有主流终端
• 原生支持中文输入、复杂快捷键、指令补全等扩展能力，适合复杂交互式服务端场景使用

方案二：纯标准库实现（无额外依赖） #

如果你的项目不允许引入第三方依赖，可以使用以下纯标准库实现的方案，自行实现同步锁、终端行控制、输入缓存逻辑：

完整实现代码 #

import threading
import random
import time
import sys
from ctypes import windll, c_int, byref

# 全局状态变量
input_buffer = []
print_lock = threading.Lock()
STD_OUTPUT_HANDLE = -11
ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004

# 初始化Windows终端ANSI转义序列支持
def init_windows_console():
    if sys.platform == "win32":
        handle = windll.kernel32.GetStdHandle(STD_OUTPUT_HANDLE)
        mode = c_int(0)
        windll.kernel32.GetConsoleMode(handle, byref(mode))
        mode.value |= ENABLE_VIRTUAL_TERMINAL_PROCESSING
        windll.kernel32.SetConsoleMode(handle, mode)

# 线程安全的打印函数
def safe_print(content):
    with print_lock:
        # 清除当前输入行
        sys.stdout.write('\033[2K\r')
        # 打印后台内容
        print(content)
        # 复原输入提示符和已输入内容
        sys.stdout.write(f"> {''.join(input_buffer)}")
        sys.stdout.flush()

# 后台模拟业务线程
def some_process():
    for i in range(50):
        time.sleep(random.randint(1, 3))
        safe_print("Some program output")

# 跨平台自定义输入函数
def custom_input(prompt="> "):
    global input_buffer
    input_buffer = []
    sys.stdout.write(prompt)
    sys.stdout.flush()
    
    while True:
        if sys.platform == "win32":
            char = windll.msvcrt.getch()
            # 处理特殊按键
            if char in (b'\x03', b'\x1a'):
                raise KeyboardInterrupt
            elif char == b'\r':
                sys.stdout.write('\n')
                res = ''.join(input_buffer)
                input_buffer = []
                return res
            elif char == b'\x08':
                if input_buffer:
                    input_buffer.pop()
                    sys.stdout.write('\b \b')
                    sys.stdout.flush()
            else:
                try:
                    decoded = char.decode('gbk')
                    input_buffer.append(decoded)
                    sys.stdout.write(decoded)
                    sys.stdout.flush()
                except:
                    pass
        else:
            # Linux/macOS 输入处理逻辑
            import tty, termios
            fd = sys.stdin.fileno()
            old_settings = termios.tcgetattr(fd)
            try:
                tty.setraw(fd)
                char = sys.stdin.read(1)
                if char == '\x03':
                    raise KeyboardInterrupt
                elif char == '\r':
                    sys.stdout.write('\n')
                    res = ''.join(input_buffer)
                    input_buffer = []
                    return res
                elif char == '\x7f':
                    if input_buffer:
                        input_buffer.pop()
                        sys.stdout.write('\b \b')
                        sys.stdout.flush()
                else:
                    input_buffer.append(char)
                    sys.stdout.write(char)
                    sys.stdout.flush()
            finally:
                termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)

if __name__ == "__main__":
    init_windows_console()
    thread = threading.Thread(target=some_process, daemon=True)
    thread.start()
    
    while True:
        try:
            shell = custom_input("> ")
            if shell.strip() == "exit":
                break
            safe_print(f"执行指令: {shell}")
        except KeyboardInterrupt:
            safe_print("\n程序退出")
            break

方案说明 #

• 兼容Windows 10 1511版本及以上，所有主流Linux发行版
• 如需适配更老的Windows版本，可将ANSI清行逻辑替换为Windows控制台API的清行操作，进一步提升兼容性

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