Python线程聊天应用：终端光标控制解决消息干扰输入问题

核心思路是利用ANSI终端控制序列配合线程锁，在收到消息时临时清空当前输入行、插入新消息，再恢复输入状态，全程不需要第三方库。

关键实现步骤 #

1. 用threading.Lock同步标准输出，避免输入线程和消息接收线程同时操作终端导致混乱。
2. 维护一个输入缓冲区，实时记录用户正在输入的内容。
3. 收到消息时，通过ANSI序列调整光标位置、清空当前输入行，打印新消息后再恢复输入提示符和已输入内容，最后把光标定位到正确位置。

完整代码示例 #

import threading
import sys
import time

# 全局变量：输入缓冲区、终端锁、输入提示符
input_buffer = []
output_lock = threading.Lock()
PROMPT = ">>> "

def receive_messages():
    """模拟消息接收线程，每隔3秒收到一条消息"""
    message_count = 0
    while True:
        time.sleep(3)
        message = f"Received: Message {message_count}"
        message_count += 1
        
        with output_lock:
            # 1. 光标移到行首，清除当前行（覆盖用户正在输入的内容）
            sys.stdout.write("\033[G\033[K")
            # 2. 打印收到的消息
            sys.stdout.write(f"{message}\n")
            # 3. 重新打印输入提示符和已输入的内容
            sys.stdout.write(f"{PROMPT}{''.join(input_buffer)}")
            # 4. 把光标定位到输入内容的末尾
            cursor_pos = len(PROMPT) + len(input_buffer)
            sys.stdout.write(f"\033[{cursor_pos}G")
            # 立即刷新输出
            sys.stdout.flush()

def input_thread():
    """处理用户输入的线程，逐字符读取并维护输入缓冲区"""
    global input_buffer
    with output_lock:
        sys.stdout.write(PROMPT)
        sys.stdout.flush()
    
    while True:
        # 逐字符读取输入，避免input()的阻塞问题
        char = sys.stdin.read(1)
        
        with output_lock:
            if char == '\n':
                # 用户按下回车，发送消息
                message = ''.join(input_buffer)
                print(f"\nSent: {message}")
                # 清空缓冲区，重新打印提示符
                input_buffer = []
                sys.stdout.write(PROMPT)
                sys.stdout.flush()
            elif char in ('\x7f', '\b'):
                # 处理退格键
                if input_buffer:
                    input_buffer.pop()
                    # 光标左移一位，清除当前字符
                    sys.stdout.write("\033[D\033[K")
                    sys.stdout.flush()
            else:
                # 普通字符，添加到缓冲区并打印
                input_buffer.append(char)
                sys.stdout.write(char)
                sys.stdout.flush()

if __name__ == "__main__":
    # 启动消息接收线程
    threading.Thread(target=receive_messages, daemon=True).start()
    # 启动输入线程
    input_thread()

代码说明 #

• \033[G：将光标移动到当前行的开头
• \033[K：清除从光标位置到行尾的所有内容
• \033[{cursor_pos}G：将光标移动到指定的列位置（从1开始计数）
• 线程锁output_lock确保所有终端输出操作都是原子性的，避免多线程下的输出混乱
• 逐字符读取输入是为了实时维护输入缓冲区，确保收到消息时能完整恢复用户的输入状态

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