搜索菜单
当前位置: 首页 > Python > Python使用pyserial读取串口数据并通过TCP发送

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

Dylan
Dylan
发布时间：2026-10-28 | 9人浏览过
原创

详细讲解如何使用 Python 的 pyserial 库读取硬件串口数据，并通过 TCP Socket 协议稳定转发至远程服务器或上位机，包含基础代码、生产级多线程与重连方案。

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

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

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

Python使用pyserial读取串口数据并通过TCP发送
图 1：串口转 TCP 数据转发整体网络拓扑与模块分工

二、环境准备与参数确认

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

1. 依赖安装

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

pip install pyserial
注意事项：在 pip 仓库中，模块名为 pyserial，请不要执行 pip install serial（后者是一个废弃的同名库），否则在代码中引入 import serial 时会引发属性缺失异常。

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）。
Python使用pyserial读取串口数据并通过TCP发送
图 2：串口读取与 TCP 转发实现四步核心流程

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

对于简单调试或数据吞吐量较低的场景，可以采用单循环结构：在同一个主循环中不断检查串口缓冲区，一旦读取到数据，立即通过 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. timeout=1.0：设置超时极为关键。如果未设置超时，调用 ser.read() 或 ser.readline() 在没有数据到来时会导致进程无限阻塞，从而无法响应退出信号或处理网络状态变化。
2. ser.in_waiting：用于查询操作系统串口接收缓冲区的现存字节数，仅在有数据时触发读取，避免频繁的阻塞等待。
3. sendall()：相较于 send()，sendall() 会持续发送直到整个字节数组全部传输完毕或发生错误，避免因网络缓冲区不足导致的部分截断丢包。

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

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

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

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

Python使用pyserial读取串口数据并通过TCP发送
图 3：多线程生产者-消费者队列与 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使用pyserial读取串口数据并通过TCP发送
图 4：串口与 TCP 网络转发核心故障排查与对策矩阵

七、总结

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

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

相关文章

相关标签: Python pyserial socket

本站声明

本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。

作者最新文章

全部内容

Python 相关产品和应用

全部下载
Python Selenium
Selenium

Selenium

4.51.0 · Windows

用于浏览器自动化和 Web 测试的 Python 工具。
Python 机器学习 Scikit-learn
scikit-learn

scikit-learn

1.9.1 · Windows

常用 Python 机器学习算法与模型评估工具库。
Python 代码质量
Black

Black

26.10.1 · Windows

强调一致性的 Python 自动代码格式化工具。
Python 代码质量
Ruff

Ruff

0.17.1 · macOS

由 Rust 编写的高速 Python 代码检查和格式化工具。
Python pytest
pytest

pytest

9.1.1 · Windows

简洁而强大的 Python 自动化测试框架。
Python FastAPI API
FastAPI

FastAPI

0.146.0 · Windows

高性能 Python API 框架，支持类型提示和自动接口文档。
Python Flask Web开发
Flask

Flask

3.1.3 · Windows

轻量灵活的 Python Web 应用框架。
Python Django Web开发
Django

Django

6.1.2 · Windows

成熟的 Python Web 框架，适合构建内容站点和业务系统。
Python IDE Spyder
Spyder

Spyder

6.1.7 · Windows

面向科学计算与数据分析的 Python 集成开发环境。
Python SciPy
SciPy

SciPy

1.18.1 · Windows

面向科学、工程和技术计算的 Python 算法库。
Python Pandas 数据分析
pandas

pandas

3.0.6 · Windows

Python 数据分析库，提供 DataFrame 与丰富的数据处理能力。
Python NumPy
NumPy

NumPy

2.5.4 · Windows

Python 科学计算基础库，提供高性能多维数组与数值运算。

文章分类

专题列表

ipython.cn 学习网
人工智能 Java AI行业服务