3、Socket实战：Python Socket服务器搭建、客户端实现、粘包处理、心跳机制设计

好，咱们直接进入正题。Socket 编程，说白了就是让两台机器能互相说话。在机器人系统里，控制器和上位机之间、控制器和传感器之间，底层通信几乎都离不开 Socket。我最早接触机器人时，用的还是串口，后来换成以太网，那感觉就像从拨号上网换成了光纤——爽快多了。

这一章，咱们手把手把 Python 的 Socket 服务器和客户端搭起来。然后解决两个最头疼的问题：粘包和心跳。嗯，这两个坑我当年都踩过，而且踩得挺深。

3.1 一个最简单的 TCP 服务器

先来个最基础的。Python 的 socket 模块是标准库，直接 import 就能用。我个人习惯把服务器封装成一个类，这样后面加功能方便。

import socket
import threading

class RobotTCPServer:
    def __init__(self, host='0.0.0.0', port=8888):
        self.host = host
        self.port = port
        self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.clients = []

    def start(self):
        self.server.bind((self.host, self.port))
        self.server.listen(5)
        print(f'[INFO] 服务器启动，监听 {self.host}:{self.port}')
        while True:
            client_sock, addr = self.server.accept()
            print(f'[INFO] 新客户端连接：{addr}')
            self.clients.append(client_sock)
            t = threading.Thread(target=self.handle_client, args=(client_sock,))
            t.start()

    def handle_client(self, sock):
        while True:
            try:
                data = sock.recv(1024)
                if not data:
                    break
                print(f'[RECV] {data.decode()}')
                # 回显，方便调试
                sock.sendall(b'ACK: ' + data)
            except:
                break
        sock.close()
        self.clients.remove(sock)
        print('[INFO] 客户端断开')

if __name__ == '__main__':
    server = RobotTCPServer()
    server.start()

这段代码里有个细节：SO_REUSEADDR。我刚开始写服务器时，经常遇到端口被占用的问题，程序一重启就报错。加上这个选项，就能立刻重用端口，省心很多。

3.2 客户端实现

客户端相对简单。但要注意，机器人控制器通常作为客户端去连接上位机，所以客户端代码要能自动重连。我在项目里遇到过控制器掉电重启后，上位机还在等它连接，结果两边都傻等的情况。

import socket
import time

class RobotTCPClient:
    def __init__(self, server_host, server_port):
        self.server_host = server_host
        self.server_port = server_port
        self.sock = None
        self.connect()

    def connect(self):
        while True:
            try:
                self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.sock.connect((self.server_host, self.server_port))
                print(f'[INFO] 已连接到服务器 {self.server_host}:{self.server_port}')
                break
            except Exception as e:
                print(f'[WARN] 连接失败，5秒后重试：{e}')
                time.sleep(5)

    def send(self, message):
        if self.sock:
            self.sock.sendall(message.encode())
            response = self.sock.recv(1024)
            return response.decode()

    def close(self):
        if self.sock:
            self.sock.close()

if __name__ == '__main__':
    client = RobotTCPClient('127.0.0.1', 8888)
    while True:
        msg = input('请输入要发送的消息：')
        if msg == 'exit':
            break
        resp = client.send(msg)
        print(f'[RESP] {resp}')
    client.close()
小提示：客户端重连时，记得每次都要新建 socket 对象。我见过有人复用同一个已关闭的 socket 去 connect，结果一直报错。嗯，这个坑我踩过。

3.3 粘包问题与解决方案

为什么会粘包？说白了，TCP 是流式协议，它不关心你一次发多少数据，只保证字节顺序正确。如果发送方连续发两条消息，接收方可能一次 recv 就把两条都读走了。这在机器人控制指令里是致命的——你发了个「前进」和「停止」，结果被当成一条指令「前进停止」，机器人就懵了。

解决粘包，常用的有三种方法：

  • 固定长度法：每条消息固定长度，不够就补零。简单粗暴，但浪费带宽。
  • 特殊分隔符法：比如用 \n 或 | 结尾。但要注意消息内容里不能出现这个分隔符。
  • 长度前缀法：先发一个固定长度的头部，告诉对方这条消息有多长。这是工业界最常用的方案。

我个人强烈推荐长度前缀法。来看代码：

import struct

def send_msg(sock, data: bytes):
    """发送带长度前缀的消息"""
    length = len(data)
    # 用4字节无符号整数表示长度
    header = struct.pack('!I', length)
    sock.sendall(header + data)

def recv_msg(sock):
    """接收带长度前缀的消息"""
    # 先收4字节头部
    header = sock.recv(4)
    if not header:
        return None
    length = struct.unpack('!I', header)[0]
    # 再收实际数据
    data = b''
    while len(data) < length:
        chunk = sock.recv(length - len(data))
        if not chunk:
            return None
        data += chunk
    return data
核心要点：接收数据时，一定要循环 recv 直到收满指定长度。因为 TCP 可能一次只收到一部分数据，尤其是网络状况不好时。我曾经在调试时发现，recv 返回的数据比预期少，就是因为没做循环接收。

3.4 心跳机制设计

心跳机制，说白了就是定期确认对方还活着。在机器人系统里，这太重要了。你想，控制器正在执行一个运动指令，突然上位机断开了，控制器如果不知道，就会一直执行下去，可能撞到人或者设备。

心跳设计有几个关键点：

  • 心跳间隔：一般 3-5 秒一次。太频繁浪费带宽，太慢无法及时检测断连。
  • 超时判定：连续 3 次没收到心跳，就认为连接断了。
  • 心跳与业务数据分离：心跳包应该用特殊标识，不要和业务数据混在一起。

我一般这样设计心跳协议：

字段 长度 说明
消息类型 1 字节 0x01 表示心跳请求，0x02 表示心跳响应
时间戳 8 字节 Unix 时间戳，用于计算延迟
保留字段 3 字节 预留，可扩展

来看心跳检测的客户端实现片段：

import threading
import time

class HeartbeatMonitor:
    def __init__(self, sock, interval=3, timeout_count=3):
        self.sock = sock
        self.interval = interval
        self.timeout_count = timeout_count
        self.missed = 0
        self.running = True

    def start(self):
        t = threading.Thread(target=self._monitor)
        t.daemon = True
        t.start()

    def _monitor(self):
        while self.running:
            time.sleep(self.interval)
            try:
                # 发送心跳请求
                heartbeat_pkt = b'\x01' + struct.pack('!Q', int(time.time())) + b'\x00\x00\x00'
                self.sock.sendall(heartbeat_pkt)
                # 这里假设有专门的接收线程处理响应
                # 如果没收到响应，missed 会由外部线程增加
            except:
                self.missed += 1

            if self.missed >= self.timeout_count:
                print('[ALERT] 心跳超时，连接可能已断开')
                self.on_timeout()
                break

    def on_timeout(self):
        """连接超时后的处理逻辑，比如触发重连"""
        self.running = False
        # 这里可以调用重连函数
注意：心跳线程不要和业务线程共用同一个 socket 的 recv 调用。否则会出现竞争条件。我建议单独开一个线程专门处理心跳，业务数据走另一个逻辑。或者用 select/poll 这种多路复用机制。

3.5 综合实战：一个带心跳的机器人控制客户端

最后，咱们把上面所有东西串起来。假设我们要做一个机器人关节控制客户端，它需要：

  1. 连接上位机服务器
  2. 发送控制指令（带长度前缀防粘包）
  3. 定期发送心跳
  4. 检测到断连后自动重连

核心结构大概是这样：

class RobotController:
    def __init__(self, server_addr):
        self.addr = server_addr
        self.sock = None
        self.heartbeat = None
        self.running = True

    def run(self):
        while self.running:
            self._connect()
            self.heartbeat = HeartbeatMonitor(self.sock)
            self.heartbeat.start()
            self._command_loop()

    def _connect(self):
        # 带重连的连接逻辑，参考前面的客户端代码
        pass

    def _command_loop(self):
        # 主循环：接收用户输入或自动发送控制指令
        while self.running:
            cmd = input('输入指令：')
            if cmd == 'quit':
                self.running = False
                break
            send_msg(self.sock, cmd.encode())
            response = recv_msg(self.sock)
            print(f'响应：{response.decode()}')

嗯，这个结构我在好几个机器人项目里都用过。实际部署时，你还需要考虑日志记录、异常上报、指令队列等。但核心骨架就是上面这些。

最后说一句：Socket 编程看似简单，但真正稳定运行起来，需要处理很多边界情况。比如网络闪断、数据包乱序、缓冲区溢出等等。我建议你在开发阶段就加入详细的日志，这样出了问题能快速定位。别问我怎么知道的——都是血泪教训。

← 上一章 📖 返回目录 下一章 →