TCP是工业/机器人领域硬件通信的主流协议（可靠、有序、无丢包），本文将以「外骨骼/数据手套的TCP通信代码」为例，拆解Python创建TCP服务器并实现硬件数据交互的完整流程，涵盖「套接字创建→连接管理→协议解析→异常处理」全环节，可直接复刻到各类硬件通信场景。

一、TCP通信核心认知

1. 核心模型：客户端-服务器（C/S）

• 服务器（Server）：被动监听指定端口，等待硬件/客户端连接，负责解析数据、处理业务；
• 客户端（Client）：主动发起连接（如硬件），发送传感器数据、接收控制指令（如振动反馈）；
• 核心特性：TCP是「面向连接、可靠传输」的协议，自动处理重传、排序，适合传感器/控制类关键数据。

2. 关键前置定义

在写核心逻辑前，需先定义通信协议相关的常量（避免硬编码，适配硬件协议）：

import socket
import struct
import numpy as np

# 1. 网络配置常量
SOCKET_LISTEN_HOST = ''  # 监听所有网卡（0.0.0.0）
DEFAULT_SOCKET_PORT = 8080  # 监听端口
SOCKET_BACKLOG = 5  # 连接等待队列长度
SOCKET_BUFFER_SIZE = 4096  # 单次接收最大字节数

# 2. 硬件协议常量（与硬件厂商约定）
PACKET_HEADER_SIZE = 2  # 包头长度（2字节）
PACKET_SIZE = 67  # 总数据包长度（包头+所有数据段）
DATA_TYPE = np.uint16  # 硬件数据类型（uint16，2字节/个）
BINARY_STRING_WIDTH = 8  # 包头二进制标志位宽度

# 3. 数据段索引常量（按硬件协议划分）
LEFT_GLOVE_START, LEFT_GLOVE_END = 2, 23  # 左手套数据段（21个值）
BACKPLATE_START, BACKPLATE_END = 23, 46  # 外骨骼数据段（23个值）
RIGHT_GLOVE_START, RIGHT_GLOVE_END = 46, 67  # 右手套数据段（21个值）

# 4. 数据转换常量
ENCODER_TO_DEGREES = 1000  # 编码器计数→角度（度）的系数
DEGREES_TO_RADIANS = np.pi / 180  # 角度→弧度转换系数
GLOVE_DOF = 21  # 手套自由度
EXOSKELETON_DOF = 23  # 外骨骼自由度

二、TCP服务器创建完整步骤

以实际项目代码中的Listener类为模板，拆解创建TCP连接的6个核心步骤：

步骤1：初始化服务器

创建类封装所有逻辑，初始化套接字、数据缓存、通信标志：

class TCPHardwareServer:
    """面向硬件的TCP服务器，解析二进制传感器数据"""
    def __init__(self, port: int = DEFAULT_SOCKET_PORT) -> None:
        # 1. 网络配置
        self.port = port
        self.server_socket = None  # 服务器套接字
        # 2. 数据缓存（存储解析后的传感器数据）
        self.q_l = np.zeros(GLOVE_DOF)  # 左手套关节弧度
        self.q_r = np.zeros(GLOVE_DOF)  # 右手套关节弧度
        self.exoskeleton_data = np.zeros(EXOSKELETON_DOF)  # 外骨骼数据
        # 3. 业务标志
        self.vibration = False  # 振动反馈使能
        self.vibe_data = np.zeros(10)  # 振动强度数据

    def normalize_angles(self, angles: np.ndarray) -> np.ndarray:
        """角度归一化（限制在[-π, π]）"""
        return (angles + np.pi) % (2 * np.pi) - np.pi

步骤2：创建并启动TCP服务器

这是服务器的「入口」，完成「创建套接字→绑定→监听→循环等待连接」：

    def start_server(self) -> None:
        """启动TCP服务器，持续监听客户端连接"""
        # 1. 创建TCP套接字（AF_INET=IPv4，SOCK_STREAM=TCP）
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        
        # 2. 绑定地址和端口（关键：HOST=''表示监听所有网卡）
        try:
            self.server_socket.bind((SOCKET_LISTEN_HOST, self.port))
        except socket.error as e:
            print(f"端口绑定失败：{e}")
            return
        
        # 3. 开始监听（BACKLOG=等待队列长度）
        self.server_socket.listen(SOCKET_BACKLOG)
        print(f"TCP服务器已启动，监听端口 {self.port}...")

        # 4. 无限循环：等待并处理客户端连接
        try:
            while True:
                # 阻塞等待客户端连接（返回：新套接字+客户端地址）
                client_socket, client_addr = self.server_socket.accept()
                print(f"已连接客户端：{client_addr}")
                
                # 处理该客户端的通信（单独逻辑）
                self.handle_client(client_socket)
        except KeyboardInterrupt:
            print("\n服务器正在关闭...")
        finally:
            # 释放资源
            if self.server_socket:
                self.server_socket.close()
                print("服务器已关闭")

步骤3：处理客户端通信

与单个客户端的持续交互，完成「接收数据→解析→业务处理→发送反馈」：

    def handle_client(self, client_socket: socket.socket) -> None:
        """处理单个客户端的通信循环"""
        try:
            while True:
                # 1. 接收数据（阻塞，直到收到数据/客户端断开）
                raw_data = client_socket.recv(SOCKET_BUFFER_SIZE)
                
                # 2. 客户端断开（无数据）
                if not raw_data:
                    print("客户端已断开连接")
                    break

                # 3. 解析数据包（核心业务）
                try:
                    device_flags = self.parse_packet(raw_data)
                    
                    # 4. 可选：发送振动反馈（反向通信）
                    if self.vibration and all(flag == '1' for flag in device_flags[:3]):
                        self.send_vibration(client_socket)
                except Exception as e:
                    print(f"数据解析错误：{e}")
                    continue
        finally:
            # 无论是否出错，关闭客户端套接字
            client_socket.close()

步骤4：解析二进制数据包

按硬件约定的协议解析数据，完成「长度校验→包头解析→数据段转换」：

    def parse_packet(self, raw_data: bytes) -> str:
        """解析硬件二进制数据包，返回设备标志位"""
        # 1. 校验数据包长度（防止丢帧）
        data_len = len(raw_data) // DATA_TYPE().nbytes
        if data_len != PACKET_SIZE:
            raise ValueError(f"数据包丢帧：期望{PACKET_SIZE}，实际{data_len}")
        
        # 2. 解析包头（2字节，小端）：获取设备标志位
        header = struct.unpack('<H', raw_data[:PACKET_HEADER_SIZE])[0]
        device_flags = bin(header)[2:].zfill(BINARY_STRING_WIDTH)  # 转二进制字符串
        
        # 3. 解析完整数据段（二进制→numpy数组，高效无拷贝）
        full_data = np.frombuffer(raw_data[:data_len * DATA_TYPE().nbytes], dtype=DATA_TYPE)
        
        # 4. 按设备标志解析对应数据
        if device_flags[0] == '1':  # 左手套数据
            self.parse_glove_data(full_data[LEFT_GLOVE_START:LEFT_GLOVE_END], is_left=True)
        if device_flags[1] == '1':  # 外骨骼数据
            self.parse_exoskeleton_data(full_data[BACKPLATE_START:BACKPLATE_END])
        if device_flags[2] == '1':  # 右手套数据
            self.parse_glove_data(full_data[RIGHT_GLOVE_START:RIGHT_GLOVE_END], is_left=False)
        
        return device_flags

    def parse_glove_data(self, raw_glove: np.ndarray, is_left: bool) -> None:
        """解析手套数据：编码器计数→角度→弧度→重排"""
        # 1. 编码器计数→角度（度）→弧度
        angles_deg = raw_glove / ENCODER_TO_DEGREES
        angles_rad = angles_deg * DEGREES_TO_RADIANS
        angles_rad = self.normalize_angles(angles_rad)
        
        # 2. 数据重排（硬件布线顺序反向，需翻转）
        finger1 = np.flip(angles_rad[:5])  # 大拇指（5关节）
        finger2 = np.flip(angles_rad[5:9])  # 食指（4关节）
        finger3 = np.flip(angles_rad[9:13]) # 中指（4关节）
        finger4 = np.flip(angles_rad[13:17])# 无名指（4关节）
        finger5 = np.flip(angles_rad[17:21])# 小指（4关节）
        parsed_data = np.hstack([finger1, finger2, finger3, finger4, finger5])
        
        # 3. 存储解析结果
        if is_left:
            self.q_l = parsed_data
        else:
            self.q_r = parsed_data

    def parse_exoskeleton_data(self, raw_exo: np.ndarray) -> None:
        """解析外骨骼数据：翻转布线顺序"""
        exo_data = raw_exo.copy()  # 转为可写数组
        # 左/右手臂数据翻转（远端→近端 → 近端→远端）
        exo_data[:9] = np.flip(exo_data[:9])   # 左手臂
        exo_data[9:18] = np.flip(exo_data[9:18]) # 右手臂
        self.exoskeleton_data = exo_data

步骤5：反向发送控制指令

TCP是双向通信，服务器可向硬件发送控制指令（如振动）：

    def send_vibration(self, client_socket: socket.socket) -> None:
        """向硬件发送振动反馈指令"""
        # 1. 构造振动数据包（按硬件协议）
        vibe_packet = np.ones(11, dtype=np.uint8) * 0
        vibe_packet[0] = 0xC0  # 标志位：0x80=左手，0x40=右手，0xC0=双手
        vibe_packet[1:11] = self.vibe_data.copy()  # 10个振动强度值
        
        # 2. 打包成二进制（小端，11个无符号字节）
        packed_data = struct.pack('<11B', *vibe_packet)
        
        # 3. 发送数据
        client_socket.sendall(packed_data)
        print(f"已发送振动指令：{vibe_packet}")
        self.vibration = False  # 重置标志

步骤6：启动服务器

if __name__ == "__main__":
    # 创建并启动TCP服务器
    server = TCPHardwareServer(port=8080)
    server.start_server()

三、关键优化与避坑指南

1. 多线程适配（避免服务器阻塞）

上述代码是单线程的，一个客户端连接会阻塞其他连接，实际场景需用线程处理客户端：

import threading

# 修改start_server中的客户端处理逻辑：
client_thread = threading.Thread(
    target=self.handle_client, 
    args=(client_socket,),
    daemon=True  # 守护线程：主线程退出时自动终止
)
client_thread.start()

2. 常见坑与解决方案

问题	原因	解决方案
端口绑定失败	端口被占用	启动前检测并释放端口（如lsof -i :8080 + kill PID）
数据包丢帧	网络抖动/硬件断连	严格校验数据包长度，解析失败时跳过该包
数据解析乱码	字节序不匹配	明确使用小端（<）/大端（>）打包（硬件通常用小端）
服务器卡死	未处理客户端断开	检测recv()返回空值时，立即关闭客户端套接字

3. 核心优化点

• 数据解析效率：用np.frombuffer直接解析二进制数据（无拷贝，比手动拆包快10倍）；
• 异常容错：所有核心逻辑加try-except，单个数据包错误不影响服务器运行；
• 资源管理：finally块确保套接字关闭，避免资源泄漏；
• 协议扩展性：常量集中定义，适配不同硬件仅需修改常量，无需改核心逻辑。

Python创建TCP服务器的6个核心步骤：

1. 前置准备：定义网络/协议常量，适配硬件通信规则；
2. 服务器初始化：封装套接字、数据缓存、业务标志；
3. 启动服务器：创建套接字→绑定端口→监听连接；
4. 客户端处理：循环接收数据，处理连接断开；
5. 协议解析：校验长度→解析包头→转换数据格式；
6. 反向通信：按需发送控制指令（如振动反馈）。