Python使用pyserial读取串口数据并通过TCP发送

在工业自动化、物联网数据采集和嵌入式开发中，将单片机、PLC或传感器的串口数据采集后通过网络上传至云端或集中监控平台是十分常见的需求。本文将基于 Python 生态中成熟的 pyserial 模块与标准网络 socket 库，直接讲解如何读取物理/虚拟串口数据，并将解析后的数据流通过 TCP 协议稳健发送至远端 TCP 服务器。

一、串口转 TCP 转发链路与核心架构

在设计串口网络转发程序前，必须清晰理解数据流经的实体与协议层级。硬件设备（如 RS-232、RS-485、USB 转 TTL 串口模块）通过底层串行通信协议将字节序列传入计算机宿主机的串口控制器中；宿主机运行的 Python 程序充当“数据网关”，一边通过 pyserial 打开对应的 COM 口或 TTY 节点进行流式读取，另一边创建 TCP Client 连接远端指定的 IP 和端口，将数据分包或原样推送到 TCP Server。

二、环境准备与参数确认

在开始编码前，必须在运行环境中安装串口支持库，并明确物理串口的核心通信参数。

1. 依赖安装

Python 标准库自带网络 socket 模块，但未内置串口硬件访问能力。我们需要安装 pyserial：

pip install pyserial

2. 确认串口名称与通信参数

串口通信双方必须严格约定以下 5 项参数，否则会导致数据乱码或完全无法接收：

• 端口号（Port）：Windows 下通常为 COM3、COM4 等；Linux 下通常为 /dev/ttyUSB0 或 /dev/ttyACM0；macOS 为 /dev/cu.usbserial-*。
• 波特率（Baudrate）：每秒传输的符号数，常见有 9600、115200、19200 等。
• 数据位（Bytesize）：通常为 8 位（serial.EIGHTBITS）。
• 停止位（Stopbits）：通常为 1 位（serial.STOPBITS_ONE）。
• 校验位（Parity）：通常为无校验（serial.PARITY_NONE）。

三、最小可行实现：直接同步转发

对于简单调试或数据吞吐量较低的场景，可以采用单循环结构：在同一个主循环中不断检查串口缓冲区，一旦读取到数据，立即通过 TCP 套接字发送出去。

import serial
import socket
import time

# 1. 配置参数
SERIAL_PORT = "COM3"      # 串口号（Linux请修改为 '/dev/ttyUSB0'）
BAUD_RATE = 115200        # 波特率
TCP_HOST = "127.0.0.1"    # TCP 服务端 IP
TCP_PORT = 9999           # TCP 服务端 端口

def main():
    # 2. 初始化串口
    try:
        ser = serial.Serial(
            port=SERIAL_PORT,
            baudrate=BAUD_RATE,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=1.0  # 读取超时设置为1秒，防止程序无限卡死
        )
        print(f"[串口] 成功打开端口: {SERIAL_PORT}, 波特率: {BAUD_RATE}")
    except serial.SerialException as e:
        print(f"[错误] 无法打开串口 {SERIAL_PORT}: {e}")
        return

    # 3. 初始化 TCP 连接
    tcp_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        tcp_client.connect((TCP_HOST, TCP_PORT))
        print(f"[网络] 成功连接至 TCP 服务端 {TCP_HOST}:{TCP_PORT}")
    except socket.error as e:
        print(f"[错误] 无法连接至 TCP 服务端: {e}")
        ser.close()
        return

    # 4. 循环读取并转发
    try:
        while True:
            # 判断串口输入缓冲区中是否有待处理数据
            if ser.in_waiting > 0:
                # 读取一行（以 \n 结尾）或读取缓冲区所有当前字节
                data = ser.readline()
                if data:
                    # 使用 sendall 确保所有字节完全发出
                    tcp_client.sendall(data)
                    print(f"[转发] 已发送 {len(data)} 字节: {data}")
            else:
                # 适度休眠，降低 CPU 空转占用
                time.sleep(0.01)

    except KeyboardInterrupt:
        print("\n[退出] 用户中断程序运行")
    except socket.error as e:
        print(f"[错误] 网络连接异常中断: {e}")
    finally:
        # 5. 确保资源安全释放
        ser.close()
        tcp_client.close()
        print("[清理] 串口与网络连接均已安全关闭")

if __name__ == "__main__":
    main()

四、生产级实现：解耦多线程与断线自愈重连

在实际生产现场中，上述单循环同步模型存在两个显著缺陷：

1. 相互阻塞与积压丢包：如果 TCP 网络遇到抖动、拥塞甚至临时断开，tcp_client.sendall() 会发生阻塞或报错，导致主循环无法继续读取串口。由于串口硬件的底层 FIFO 缓冲区非常有限（通常只有几十字节到几 KB），一旦 Python 没能及时排空，硬件缓冲区就会溢出（Buffer Overflow），造成不可挽回的数据丢失。
2. 缺乏容错恢复：一旦网络瞬断，程序直接崩溃退出，无法无人值守长期稳定运行。

针对这两个问题，业界成熟的解决方案是生产者-消费者模型：独立的串口读取线程负责将收到的数据迅速写入内存队列 queue.Queue，而 TCP 发送线程专门从队列中消费数据并通过网络发送。当网络断开时，发送线程自动执行重试和重连，而串口读取线程不受任何影响继续采集。

工业级多线程与断线重连完整代码

import serial
import socket
import threading
import queue
import time
import sys

# 通信配置
SERIAL_PORT = "COM3"
BAUD_RATE = 115200
TCP_HOST = "127.0.0.1"
TCP_PORT = 9999

# 全局数据缓冲队列（容量限制防止极端网络中断下内存泄露）
data_queue = queue.Queue(maxsize=5000)
# 停止运行控制标志
stop_event = threading.Event()

def serial_worker():
    """串口采集工作线程：从硬件持续读取数据写入队列"""
    print("[串口线程] 正在尝试打开串口...")
    try:
        ser = serial.Serial(
            port=SERIAL_PORT,
            baudrate=BAUD_RATE,
            timeout=0.5
        )
        print(f"[串口线程] 串口 {SERIAL_PORT} 已就绪")
    except serial.SerialException as e:
        print(f"[串口线程错误] 打开失败: {e}")
        stop_event.set()
        return

    while not stop_event.is_set():
        try:
            if ser.in_waiting > 0:
                raw_bytes = ser.read(ser.in_waiting)
                if raw_bytes:
                    try:
                        # 队列满时丢弃最旧数据或报警（此处设置0.1s超时放入）
                        data_queue.put(raw_bytes, timeout=0.1)
                    except queue.Full:
                        print("[警告] 缓冲队列已满，丢弃历史数据包")
            else:
                time.sleep(0.005)
        except serial.SerialException as e:
            print(f"[串口线程错误] 串口通信异常中断: {e}")
            break

    ser.close()
    print("[串口线程] 串口已关闭")

def tcp_worker():
    """TCP 传输工作线程：从队列消费数据并支持断网自愈重连"""
    sock = None

    def get_connected_socket():
        """尝试建立 TCP 连接，失败时循环重试"""
        while not stop_event.is_set():
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                # 设置发送超时，防止网络死锁
                s.settimeout(5.0)
                s.connect((TCP_HOST, TCP_PORT))
                print(f"[TCP 线程] 成功连接至服务端 {TCP_HOST}:{TCP_PORT}")
                return s
            except (socket.error, socket.timeout) as e:
                s.close()
                print(f"[TCP 连接等待] 无法连通服务端 ({e})，3秒后重试...")
                time.sleep(3.0)
        return None

    sock = get_connected_socket()

    while not stop_event.is_set():
        if sock is None:
            sock = get_connected_socket()
            if sock is None:
                break

        try:
            # 阻塞1秒等待数据，若无数据超时则继续检查循环
            data = data_queue.get(timeout=1.0)
        except queue.Empty:
            continue

        try:
            sock.sendall(data)
            data_queue.task_done()
        except (socket.error, socket.timeout) as e:
            print(f"[TCP 传输异常] 发送失败: {e}，准备重新连接")
            try:
                sock.close()
            except Exception:
                pass
            sock = None
            # 将未能发出的数据重新塞回队列头（或保留在重连后发送）
            try:
                data_queue.put(data, block=False)
            except queue.Full:
                pass

    if sock:
        sock.close()
    print("[TCP 线程] 网络连接已安全注销")

if __name__ == "__main__":
    t_serial = threading.Thread(target=serial_worker, name="SerialThread")
    t_tcp = threading.Thread(target=tcp_worker, name="TcpThread")

    t_serial.daemon = True
    t_tcp.daemon = True

    t_serial.start()
    t_tcp.start()

    print("[主进程] 数据采集转发服务已启动，按 Ctrl+C 安全停止")
    try:
        while not stop_event.is_set():
            time.sleep(0.5)
    except KeyboardInterrupt:
        print("\n[主进程] 收到终止信号，正在协调子线程退出...")
        stop_event.set()

    t_serial.join(timeout=2.0)
    t_tcp.join(timeout=2.0)
    print("[主进程] 服务已彻底终止")

五、实战调试与联调验证

在没有物理单片机或远程服务器时，可以通过软件模拟手段完整验证程序的正确性。

1. 模拟 TCP 服务端

在终端中，可以使用系统自带的 nc（netcat）工具或简单的 Python 脚本充当 TCP 服务端进行监听：

# Linux / macOS 终端
nc -l -p 9999

# 或者使用 Python 快速启动单行 TCP 服务端输出接收到的数据：
python -c "import socket; s=socket.socket(); s.bind(('0.0.0.0', 9999)); s.listen(1); c,a=s.accept(); print('连接来自', a); [print(c.recv(1024).decode(errors='replace'), end='') for _ in iter(int, 1)]"

2. 模拟虚拟串口（Loopback）

• Linux / macOS：可以使用 socat 命令创建成对的虚拟 TTY 终端：

  socat -d -d pty,raw,echo=0 pty,raw,echo=0

  命令会输出两个节点，例如 /dev/pts/1 与 /dev/pts/2。Python 读取 /dev/pts/1，你在另一个终端向 /dev/pts/2 输入 echo "Hello Sensor" > /dev/pts/2 即可触发转发。
• Windows：可以安装成熟的开源或免费虚拟串口驱动工具（例如 com0com 或 VSPD），将虚拟的 COM1 与 COM2 绑定成对。

六、常见问题与排查诊断矩阵

串口通信与 Socket 编程横跨硬件驱动与操作系统网络协议栈，初学者经常遇到权限受限、乱码、粘包等经典问题。下表列出核心故障与针对性对策：

七、总结

通过 Python 搭建串口至 TCP 的数据传输通道，核心逻辑在于清晰的模块边界与合理的吞吐解耦：

• 在串口侧，使用 pyserial 严格校准物理层参数，并配置合理的 timeout，避免无数据时的永久阻塞；
• 在网络侧，使用 sendall() 保证字节流的完整性，并必须明确应用层协议（如行分割或帧头帧尾）以应对 TCP 字节流粘包现象；
• 在面对复杂现场环境时，切忌使用单线程直接串联两者，而应引入基于内存队列的多线程并发架构，彻底隔离网络波动对高频串口采集的影响，保证数据链路的长期健壮稳定。