如何用Python实现客户端键盘输入实时同步至TCP服务器

核心逻辑 #

• 服务器：创建TCP套接字监听连接，持续接收客户端发送的单字节数据，实时输出到终端（处理退格字符实现删除效果）
• 客户端：修改终端输入模式，实现无回车实时捕获单个字符，将每个字符（含退格）发送给服务器，同时本地同步显示

服务器端代码 #

import socket

def main():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(('localhost', 12345))
    server_socket.listen(1)
    print("等待客户端连接...")
    
    client_conn, client_addr = server_socket.accept()
    print(f"已连接客户端：{client_addr}")
    
    try:
        while True:
            data = client_conn.recv(1).decode('utf-8')
            if not data:
                break
            
            # 兼容Linux和Windows的退格字符
            if data == '\x7f' or data == '\b':
                # 通过退格+空格+退格实现终端字符删除
                print('\b \b', end='', flush=True)
            else:
                print(data, end='', flush=True)
    except KeyboardInterrupt:
        print("\n服务器已停止")
    finally:
        client_conn.close()
        server_socket.close()

if __name__ == "__main__":
    main()

客户端代码 #

import socket
import sys
import os

def set_terminal_raw():
    # 配置终端为非规范模式（无需回车获取字符）
    if os.name == 'posix':
        import termios
        import tty
        old_attrs = termios.tcgetattr(sys.stdin)
        try:
            tty.setraw(sys.stdin.fileno())
            return old_attrs
        except:
            termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_attrs)
            raise
    elif os.name == 'nt':
        import msvcrt
        return None
    else:
        raise NotImplementedError("不支持当前操作系统")

def restore_terminal(old_attrs):
    # 恢复终端原有属性，避免终端异常
    if os.name == 'posix' and old_attrs is not None:
        import termios
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_attrs)

def main():
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        client_socket.connect(('localhost', 12345))
        print("已连接服务器，开始输入（Ctrl+C退出）：")
        
        old_attrs = set_terminal_raw()
        try:
            while True:
                if os.name == 'posix':
                    char = sys.stdin.read(1)
                elif os.name == 'nt':
                    char = msvcrt.getch().decode('utf-8')
                
                # 捕获Ctrl+C退出信号
                if ord(char) == 3:
                    break
                
                client_socket.send(char.encode('utf-8'))
                
                # 本地同步显示输入内容
                if char == '\x7f' or char == '\b':
                    print('\b \b', end='', flush=True)
                else:
                    print(char, end='', flush=True)
        finally:
            restore_terminal(old_attrs)
    except ConnectionRefusedError:
        print("无法连接到服务器，请确保服务器已启动")
    except KeyboardInterrupt:
        print("\n客户端已退出")
    finally:
        client_socket.close()

if __name__ == "__main__":
    main()

关键细节说明 #

• 终端模式修改：客户端通过set_terminal_raw禁用终端的规范输入模式，实现无需回车即可获取单个字符，退出时恢复终端属性避免异常
• 退格处理：通过输出\b \b组合序列，实现终端上的字符删除效果，兼容Linux（\x7f）和Windows（\b）的退格字符
• 实时刷新：使用flush=True强制刷新输出缓冲区，确保字符实时显示，避免因缓冲区缓存导致的延迟

使用步骤 #

1. 先运行服务器端代码
2. 启动客户端代码
3. 在客户端终端输入字符，服务器端会实时同步显示；按退格键时，服务器端同步删除对应字符
4. 按Ctrl+C可退出客户端和服务器

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