Python如何在input()接收用户输入时同步执行print()打印

input()本身是阻塞式的终端输入调用，默认终端行缓冲模式下，你当前的多线程写法会直接触发IO冲突：后台线程收到新消息调用print()时，会直接把内容插在用户正在输入的行位置，打乱输入提示符和用户已经输入的内容。
除此之外你的现有代码还有3个明显逻辑问题：

• 每次循环都新建、销毁两个线程，消息检查逻辑只执行一次就退出，根本做不到持续监听新消息
• 用eval()解析接口返回内容属于严重安全漏洞，接口返回内容一旦被篡改，会直接在本地执行任意恶意代码
• 全局Message对象没有加线程锁保护，多线程同时读写很容易出现数据错乱

核心处理逻辑是：后台线程需要打印新消息时，先暂存用户当前已经输入的内容，清空当前终端行、打印完新消息后，再把输入提示符和用户已输入的内容重新打印到当前行，完全不打断输入流程。
完整可运行代码如下：

import requests
import threading
import time
import readline
from threading import Lock

# 全局状态初始化
class User:
    username = "your_username"  # 替换为你自己的用户名赋值逻辑
class Message:
    author = ""
    content = ""

msg_lock = Lock()
INPUT_PROMPT = "> "

def print_incoming_msg(author: str, content: str) -> None:
    """打印新消息时不打断用户正在输入的内容"""
    # 读取当前输入缓冲区用户已经输入的内容
    current_input = readline.get_line_buffer()
    # 光标移到行首，清空整行内容
    print(f"\r\033[K{author}: {content}")
    # 重新输出提示符和用户已输入的内容，不刷新换行
    print(f"{INPUT_PROMPT}{current_input}", end="", flush=True)

def listen_new_msg() -> None:
    """后台常驻线程，持续轮询新消息"""
    get_url = "your_new_msg_api_url"  # 替换为你的消息拉取接口地址
    last_author = ""
    last_content = ""
    while True:
        try:
            # 用requests自带json解析，绝对不要用eval
            resp = requests.get(get_url, timeout=5).json()
            auth = resp.get("auth", "")
            cont = resp.get("cont", "")
            # 跳过自己发的消息、已经打印过的重复消息
            if auth != User.username and (auth != last_author or cont != last_content):
                with msg_lock:
                    Message.author = auth
                    Message.content = cont
                    last_author = auth
                    last_content = cont
                print_incoming_msg(auth, cont)
        except Exception:
            # 接口异常时延长重试间隔
            time.sleep(2)
        # 轮询间隔，根据需求调整，不建议小于0.3秒
        time.sleep(0.5)

def main():
    post_url = "your_send_msg_api_url"  # 替换为你的消息发送接口地址
    # 启动后台消息监听线程，设为守护线程随主进程退出
    threading.Thread(target=listen_new_msg, daemon=True).start()

    while True:
        input_content = input(INPUT_PROMPT).strip()
        if not input_content:
            continue
        # 加锁更新全局消息状态
        with msg_lock:
            Message.author = User.username
            Message.content = input_content
        print(f"{User.username}: {input_content}")
        # 发送消息到服务端
        try:
            requests.post(post_url, data={"auth": User.username, "cont": input_content}, timeout=5)
        except Exception:
            print_incoming_msg("系统", "消息发送失败，请检查网络连接")

if __name__ == "__main__":
    main()

• 代码里用到的\r\033[K是标准ANSI终端转义序列，Windows 10及以上、所有主流Linux/macOS终端都默认支持，不需要额外安装第三方依赖，比curses等跨终端方案轻量很多。
• 轮询间隔建议设置在0.3-1秒区间，间隔太短会给服务端造成不必要的请求压力，间隔太长会提升消息接收延迟。
• 如果需要兼容更老的Windows终端，可以把代码里的ANSI清行逻辑替换为调用Windows API清空控制台行，逻辑本质是一样的。

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