Python实现串口与TCP互相转发:原理、代码与实战
发布时间:2026/10/2 10:45:01来源:尧图网络
做串口和TCP互相转发这件事乍一听好像挺简单——不就是把串口收到的字节塞进TCP socket再把socket收到的字节丢回串口嘛。但真到项目里跑起来你会发现在Windows上随便找个串口助手就能干的活放到Linux服务器上、放进自动化测试脚本里、或者塞进一个需要7x24小时跑的设备中控程序里一下子就会冒出无数细节问题串口被占用怎么办、TCP断线要不要重连、数据粘包半包怎么处理、读写线程怎么协作才不会丢数据、CH340驱动和USB转串口芯片在Linux底下识别名怎么老变……这篇文章我就围绕“串口和TCP互相转发工具”这个主题把我自己从零手写这类工具、以及使用socat、Python、串口助手组合方案时踩过的坑和梳理出的思路一次讲清楚。1. 项目背景为什么需要把串口接到TCP上1.1 不只是“串口联网”而是调试方式的升级如果你一直做单片机、PLC、工业控制或者物联网设备开发下面这些场景你一定不陌生设备只有RS232/RS485串口但机房和办公网早就没有串口线了想远程看日志只能跑一趟现场。一块单片机通过串口和一个上位机软件通讯你想在中间加一层转发把串口报文同时推送给调试工具和数据库采集服务。用Modbus RTU从站设备想通过网口被Modbus TCP主站轮询需要做协议封装转换。C51这类老单片机做串口IAP升级时维护人员希望通过网络远程把固件推下去避免去现场开盖接串口线。这些需求归根到底都是一件事把两个不同物理链路上的字节流打通。串口是本地、低速、按字节收发的物理链路TCP是网络、流式、面向连接的逻辑链路。中间加一个“转换器”数据从串口进去变成TCP包出来从TCP进去变成串口字节流出来就是一个最朴素的“转发工具”。1.2 硬件网关与软件转发怎么选市面上的串口转以太网硬件模块比如USR-TCP232、有人物联网的串口服务器、工业现场常见的串口转WiFi模块本质上就是一个“嵌入式串口-网口双向桥”它们适配工业现场的长期部署稳定、小巧、上电即用。但我个人在大多数开发调试场景里反而更偏爱纯软件的转发方案原因有三点硬件模块虽然方便但很多型号对自定义波特率、停止位、流控的支持有限出了问题不好改固件。开发阶段需要频繁改配置、看实时日志、甚至自动跑回归软件方案可以无缝嵌入脚本。软件可以在同一台电脑上虚拟多个串口、多个端口灵活做“一对多”分发硬件模块做这个事就很费劲。所以如果你只是在现场长期运行某一条产线硬件串口服务器是优选如果是在实验室做开发调试、写自动化测试、做远程维护的跳板我强烈建议搞一个软件转发工具并且最好是自己能掌控源码的那种。1.3 这个工具到底适合谁嵌入式软件工程师、工控上位机开发、测试工程师、运维人员、物联网爱好者都可能用到这个工具。它会伴随你解决这些问题远程访问一台只有串口接口的Linux开发板不再需要物理串口线。抓取串口通讯报文同时转发给多个TCP客户端做监听分析。把串口调试助手变为“网络版本”让TCP客户端也能直接和设备对话。在自动化测试框架里用Python直接和串口设备交互中间不需要人肉点“发送”。这篇文章后面给到的实现方案基于Python和pyserial你完全可以拿它当一个轻量级的中转程序也能在它的基础上扩展出帧解析、日志保存、多客户端管理等功能。2. 转发工具的核心设计思路2.1 单向转发 vs 双向桥接最简单的转发工具只会做一件事从串口读到数据通过网络发送出去或者从网络收到数据往串口写。可一旦要“互相转发”就必须考虑双向同时工作。串口设备和远端TCP客户端可能同时都在发数据如果代码把读写都放在一个线程里串行处理非常容易因为某一端的阻塞拖死另一端。我在实际开发中的处理原则是一个方向一条独立的转发通道互不等待。也就是“串口读取线程”负责把串口数据搬到TCP发送缓冲区“TCP读取线程”负责把网络数据搬到串口发送缓冲区两个线程完全独立。即使某一段时间TCP网络卡住串口读取线程也不应该停止从串口取数据否则设备端的发送缓冲区会满进而把设备卡死。2.2 TCP服务端还是客户端互转工具里的TCP侧一定要支持两种角色TCP Server模式工具监听一个本地端口等待别的设备/软件来连接。常用于你本地调试环境需要多个客户端同时连上来收数据的时候比如一边开着网络调试助手一边跑着自己的采集脚本。TCP Client模式工具主动连接远端的TCP服务端。常用于把串口设备的数据推送到机房服务器或者连接到PLC、远程上位机。我一开始写这类工具时只做了Server模式觉得这样用起来方便。但后来在工控现场发现很多上位机软件是固定的TCP服务端设备数据需要主动连上去没有Client模式根本行不通。所以如果你要复现这类工具两个角色一定要都做。2.3 串口参数不是“默认就行”串口转发遇到“乱码”或者“数据总是读不全”时绝大多数情况不是工具的问题而是串口参数对不上。一个完整的串口配置包含以下参数波特率比如9600、115200、460800。通信双方不一致就直接乱码。数据位一般8位早期单片机也有7位数据位。停止位1位或2位。停止位不对接收方容易出现帧错误。校验位None/Odd/Even。尤其Modbus RTU设备常配置为无校验或偶校验选错了数据就废了。流控硬件流控RTS/CTS、软件流控XON/XOFF。如果设备开启了流控而转发工具没开数据量大时大概率丢字节。另外一个高频问题就是“USB转串口”的驱动。Windows下CH340、CP2102、FT232这些芯片如果不装驱动设备管理器里看到的只会是一个带问号的“未知设备”工具自然打不开串口。Linux下虽然没有安装驱动的问题但设备名会受插入顺序影响今天是ttyUSB0明天可能变成ttyUSB1写死设备名容易被坑。2.4 软件转发和硬件网关的选型差异很多人会问既然有现成硬件模块为什么还要写软件这里我把两者的差异做了个对比对比项软件转发工具硬件串口服务器部署位置电脑、开发板、服务器工业现场、设备柜内配置灵活性高随时改代码改参数中需通过网页/AT指令配置调试友好度高可打印日志、可加断点低黑盒运行运行稳定性依赖宿主系统高专用硬件相对稳定多客户端支持容易实现看具体型号价格免费用Python/开源工具几十到几百元不等从我个人的体感来看两者并不是替代关系。硬件网关适合长期运行的生产环境软件工具适合研发、测试、临时排查。在项目前期用软件把通讯逻辑调通再部署硬件设备到现场是我比较推荐的节奏。3. 工具选型用现成的还是自己写3.1 Linux生态里常用方案如果你只是临时需要把串口和TCP打通不想写代码Linux环境下有几个现成方案网络工具ncat它是Nmap套件里的工具支持--sh-exec这种把socket数据交给外部命令处理的能力可以搭配cat管道和串口设备做互转但用法比较绕。socat这是我最常用的方案它的核心思想就是“把两个数据通道对接”语法简单支持TCP、UDP、PTY、文件、串口等多种通道类型。举个例子把串口/dev/ttyUSB0接到TCP端口8888上可以这样socat -d -d /dev/ttyUSB0,raw,nonblock,b115200,cs8,parenb0,stopb1 TCP-LISTEN:8888,reuseaddr这条命令会创建一个TCP Server把ttyUSB0串口数据和TCP端口8888双向对接。-d -d是打印调试信息raw表示原始透传nonblock避免串口阻塞b115200是波特率cs8是8位数据位parenb0是无校验stopb1是1位停止位。光看这一条命令就知道socat把参数全部堆在命令行里适合临时调试但配置多了以后可读性很差。而且它默认的缓冲区策略和异常重连机制有时候不够用比如串口设备忽然断开后socat进程并不会尝试自动重新打开串口。3.2 Windows下常用的串口助手类工具Windows上大家用得多的就是各种串口调试助手比如XCOM、友善串口助手、Commix等。这些工具的优点是界面直观、即点即用适合人肉调试但“互相转发”的能力却参差不齐有些串口助手虽然有“网络透传”功能但只支持TCP Server模式不支持主动连接。很多工具不能同时开两个转发的实例换个参数就得重启软件。想把这些工具嵌入自动化脚本基本没戏只能靠按键精灵之类的方案去模拟界面操作维护成本很高。所以对于需要长期运行、自动化、可配置化的场景我还是倾向于自己写一个Python工具。这样既能复用pyserial来管理串口也能用socket灵活处理TCP代码在自己手里出了问题随时能改。3.3 为什么最终选择了Python方案我做这个“串口和TCP互相转发工具”时最初也是先用socat跑通验证链路但最后落地时选择了Python理由很明确跨平台Windows和Linux下都能跑。实验室用Windows笔记本部署到树莓派、工控机时用LinuxPython代码不需要改动逻辑。pyserial管串口很成熟枚举串口、设置波特率、读写超时、流控这些都能控制而且能根据系统自动解析设备名。扩展性强可以在转发过程中插入报文解析、CRC校验、日志记录、断线重连、心跳保活等功能。调试方便Python能快速验证想法报错信息也直观配合logging模块可以打印完整的收发日志。当然缺点也有性能不如C/C或Go实现极端高并发场景下会吃亏。但串口本身速率也就几百KbpsTCP转发只是搬运字节流Python完全够用。如果你的场景是大量并发的网络连接那就另选技术栈了。4. 实操从零实现一个串口-TCP互转工具4.1 环境和依赖准备我用的环境是Python 3.8依赖只有一个pyserialsocket是标准库自带。安装很简单pip install pyserial在Windows下如果串口被识别为COM3这种名字直接写COM3就行。在Linux下先用ls /dev/ttyUSB*或ls /dev/ttyACM*确认设备名然后配置为/dev/ttyUSB0。有一点要特别注意Linux下访问串口需要权限如果你不想每次都用sudo运行可以把当前用户加到dialout组sudo usermod -a -G dialout $USER加完组以后重新登录终端再运行程序就不会报“Permission denied”了。4.2 用配置文件管理参数写工具的第一步不是直接写逻辑而是先想好配置怎么组织。我习惯用JSON文件管理所有参数程序每次启动时读取这样改参数不用动代码。配置项是这样设计的{ serial: { port: COM3, baudrate: 115200, bytesize: 8, parity: N, stopbits: 1, timeout: 0.1, xonxoff: false, rtscts: false }, tcp: { mode: server, host: 0.0.0.0, port: 8888, remote_host: 192.168.1.100, remote_port: 9000, reconnect_interval: 3 }, log: { level: INFO, file: forward.log } }这里tcp.mode有两个值server表示工具监听本地端口client表示工具主动连接远端服务器。serial.timeout设置串口读取超时时间不能设得太长否则串口线程会卡住导致实时性变差一般0.05到0.2秒就够了。tcp.reconnect_interval是TCP Client模式下的自动重连间隔这个参数对长期稳定运行特别重要。4.3 完整代码实现这里我给出一个精简但可用的完整实现代码结构分成三个文件config.json配置文件。serial_forward.py主程序。先看主程序代码我按照功能模块拆开讲解。import json import threading import socket import serial import logging import sys import time logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s) logger logging.getLogger(__name__) def load_config(pathconfig.json): with open(path, r, encodingutf-8) as f: cfg json.load(f) return cfg def open_serial(cfg_serial): port cfg_serial[port] ser serial.Serial() ser.port port ser.baudrate cfg_serial[baudrate] ser.bytesize cfg_serial[bytesize] ser.parity cfg_serial[parity] ser.stopbits cfg_serial[stopbits] ser.timeout cfg_serial[timeout] ser.xonxoff cfg_serial.get(xonxoff, False) ser.rtscts cfg_serial.get(rtscts, False) ser.open() logger.info(f串口已打开: {port} {ser.baudrate}) return ser这个函数没什么玄机就是把配置映射为serial.Serial对象的属性。我特意把xonxoff和rtscts都暴露出来因为有的设备会用到软件流控或硬件流控默认关闭。如果你在数据量大时高频丢字节可以尝试把rtscts打开试试。def create_tcp_server(host, port): srv_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) srv_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) srv_sock.bind((host, port)) srv_sock.listen(5) logger.info(fTCP Server已启动: {host}:{port}) return srv_sock def create_tcp_client(host, port): sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((host, port)) logger.info(fTCP Client已连接: {host}:{port}) return sock创建TCP Server和Client的逻辑也是常规操作。这里有个细节Server模式下我用SO_REUSEADDR避免程序退出后端口处于TIME_WAIT状态导致重启报“bind: only one usage of each socket address”。这个问题在Windows和Linux上都会出现特别是反复调试、频繁重启工具时。如果你不设置这个选项很可能遇到error: listen tcp 127.0.0.1:8888: bind: only one usage of each socket addre在客户端模式下connect失败会抛异常所以必须放在重试循环里。def serial_to_tcp(ser, sock_getter, stop_event): while not stop_event.is_set(): try: data ser.read(ser.in_waiting or 1) if data: sock sock_getter() if sock is not None: sock.sendall(data) except (serial.SerialException, OSError) as e: logger.error(f串口读取/网络发送异常: {e}) time.sleep(0.5) except Exception as e: logger.error(f未知异常: {e}) break串口到TCP的转发思路很简单每次读取串口缓冲区内已有的数据如果有数据就通过TCP sendall发送。ser.read(ser.in_waiting or 1)的意思是如果in_waiting不为0就读取这么多字节如果为0就调用read(1)阻塞等待1个字节并在timeout时间内超时返回。这种写法能兼顾实时性和CPU占用。这里关键的是sock_getter函数它返回当前可用的TCP socket。在Server模式下客户端可能会断开socket对象会变化在Client模式下socket需要重连。所以我把它做成一个可调用的函数每次发送前动态获取最新的socket避免多线程访问同一个失效socket。def tcp_to_serial(ser, sock_getter, stop_event): while not stop_event.is_set(): sock sock_getter() if sock is None: time.sleep(0.2) continue try: data sock.recv(4096) if data: ser.write(data) else: logger.info(TCP连接对方已关闭连接) if isinstance(sock_getter(), socket.socket): sock_getter().close() time.sleep(0.2) except (socket.timeout, ConnectionError, OSError) as e: logger.warning(f网络读取异常: {e}) time.sleep(0.5)TCP到串口的转发原理也一样只不过read变成了recv。这里要注意一点recv(4096)返回空字节串表示对端关闭连接。如果检测到这种情况就不要傻乎乎地继续用这个socket了应当触发重连或者等待新客户端接入。def tcp_server_loop(tcp_cfg, sock_holder, stop_event): srv_sock create_tcp_server(tcp_cfg[host], tcp_cfg[port]) while not stop_event.is_set(): try: client_sock, addr srv_sock.accept() logger.info(f新的TCP客户端接入: {addr}) sock_holder[sock] client_sock except OSError as e: logger.error(faccept异常: {e}) time.sleep(1)Server模式下我把当前连接的客户端socket放在一个共享字典sock_holder里。这个函数的线程只负责接收新客户端把新连接替换旧连接。注意这里我采用“后接入的连接顶掉旧连接”的策略在大多数调试场景下这是合理的因为调试工具有时候会反复重连旧连接反正也失效了。def tcp_client_loop(tcp_cfg, sock_holder, stop_event): while not stop_event.is_set(): try: sock create_tcp_client(tcp_cfg[remote_host], tcp_cfg[remote_port]) sock_holder[sock] sock sock_holder[sock].settimeout(1.0) except Exception as e: logger.error(fTCP连接失败: {e}, {tcp_cfg[reconnect_interval]}秒后重试) time.sleep(tcp_cfg[reconnect_interval]) continue while not stop_event.is_set(): time.sleep(0.5) if sock_holder[sock] is None: break try: sock_holder[sock].close() except Exception: pass sock_holder[sock] None time.sleep(tcp_cfg[reconnect_interval])Client模式比Server模式麻烦一点因为连接断开后要自动重连。我这里的策略是主循环里先尝试连接成功以后进入内层循环“保持连接”每隔0.5秒检查一次socket是否已被上层标记为失效。如果TCP转发线程发现对端关闭会把sock_holder[sock]置为None这时外层循环就会重新连接。实际项目中你还可以给这个循环加一个最大重试次数防止服务端永久不恢复时长尾连接。def main(): cfg load_config() serial_cfg cfg[serial] tcp_cfg cfg[tcp] stop_event threading.Event() sock_holder {sock: None} try: ser open_serial(serial_cfg) except Exception as e: logger.error(f串口打开失败: {e}) sys.exit(1) def current_sock(): return sock_holder[sock] if tcp_cfg[mode] server: t threading.Thread(targettcp_server_loop, args(tcp_cfg, sock_holder, stop_event), daemonTrue) t.start() else: t threading.Thread(targettcp_client_loop, args(tcp_cfg, sock_holder, stop_event), daemonTrue) t.start() t1 threading.Thread(targetserial_to_tcp, args(ser, current_sock, stop_event), daemonTrue) t2 threading.Thread(targettcp_to_serial, args(ser, current_sock, stop_event), daemonTrue) t1.start() t2.start() logger.info(转发工具已启动按CtrlC退出) try: while True: time.sleep(1) except KeyboardInterrupt: logger.info(收到退出信号正在停止...) stop_event.set() try: sock_holder[sock].close() except Exception: pass ser.close() sys.exit(0) if __name__ __main__: main()主函数的逻辑很清晰打开串口启动TCP监听/连接线程再启动两个方向的转发线程然后进入主循环等待退出信号。用threading.Event作为全局停止标记所有线程在退出时都能安全结束。4.4 运行和验证方法代码写完后先用最简单的方式验证。在终端运行python serial_forward.py看到日志输出“串口已打开”和“TCP Server已启动”之后打开一个TCP调试助手Windows下可以用SocketToolLinux下可以用nc连接127.0.0.1:8888。然后用另一台设备或虚拟串口往COM3发数据比如发“hello serial”在TCP调试助手里应该能看到同样的字节。反向测试更简单在TCP调试助手里输入“hello tcp”用串口助手的接收区应该能看到数据并且可以接一个USB转TTL的回路测试TX接RX直接验证串口侧是否真的发出去了。4.5 关于“串口烧写失败”和“单片机串口升级”的联动说个实际场景我最开始写这个工具就是受“C51单片机串口升级架构”的启发。老式的C51串口IAP升级大致是单片机里固化一段Bootloader上位机通过串口把bin文件分包发下去Bootloader收到后写入Flash。这个过程对时序很敏感如果中间加一个网络转发TCP的延迟抖动、粘包拆包都会影响升级成功率。所以如果你的目标是“远程串口烧写”给你几个实用的建议在TCP侧启用Nagle算法禁用TCP_NODELAY避免小包被延迟合并。我的示例代码里没有加你可以在socket创建后加一行sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)。在TCP对端加一个简单的“应答重传”逻辑烧写协议本身最好能识别“没收到应答就重发当前包”否则断网重连后状态很难恢复。转发的缓冲区要足够大。串口一次写入Flash的bin包可能很大缓冲区太小在TCP链路抖动时容易丢数据。5. 常见问题与排查技巧5.1 启动失败类问题速查现象可能原因解决办法打开串口报“找不到指定端口”串口被其他软件占用或设备名写错先关掉所有串口调试助手Windows查看设备管理器确认COM号Linux执行ls /dev/ttyUSB*Linux打开串口报“Permission denied”当前用户不在dialout组sudo usermod -a -G dialout $USER重新登录TCP Server启动报“Address already in use”端口被其他进程占用或上一次程序未完全退出设置SO_REUSEADDR使用netstat -ano找到占用进程杀掉Windows上串口识别为“未知设备”CH340/CP2102/FT232驱动没装去官网下载对应驱动安装CH340常见芯片型号可用“CH340驱动”搜索官方驱动包5.2 转发过程中的数据异常下面这几个问题是我被坑得最多的地方数据乱码检查串口双方的波特率、数据位、停止位、校验位是否一致。特别是Modbus RTU设备有的默认是8N1有的是8E1参数不匹配时收到的数据就是一堆无意义的十六进制。可以用示波器或逻辑分析仪看电平确认到底是工具配置问题还是物理链路问题。能收到前几个字节后面就没了很可能是缓冲区溢出。默认的串口接收缓冲区是4096字节TCP接收缓冲区几十KB但如果串口读取线程不能及时取走数据高波特率下数据会丢。把serial.timeout调小到0.05秒增加读取频率同时适当增大ser.set_buffer_size(rx_size65536, tx_size65536)。数据粘连粘包TCP是流式协议应用程序send两次的数据对端recv一次可能全收到也可能被拆成多段。这不属于转发工具的bug但如果你对端是单片机Bootloader一定要在自己的协议层处理分包比如在bin文件包前面加固定的帧头、包长度字段和CRC校验接收方按长度字段解析。串口到TCP方向一般不需要处理粘包但TCP到串口方向必须考虑。数据丢一半检查串口是否启用了流控。RS232设备在数据量稍大时如果启用了硬件流控而你的工具没开rtscts对端会一直等CTS信号导致发送方认为链路阻塞从而丢数据。反过来如果你的设备没接流控线工具却开了RTS/CTS也会莫名其妙丢字节。5.3 TCP连接相关的高频问题TCP连接涉及到三次握手、四次挥手、长连接和短连接这些概念转发工具在这个层面最常遇到的问题基本集中在三个方面客户端连接不上Server模式端口先确认防火墙。CentOS上如果安装了firewalld默认是不放行自定义端口的需要firewall-cmd --zonepublic --add-port8888/tcp --permanent然后firewall-cmd --reload。Windows下则是“Windows Defender防火墙”里是否允许Python程序监听端口。Client模式连不上远端服务端用telnet ip port先测——注意我的意思不是让你用telnet连本工具而是先确认远端服务端端口可达。如果远端是云服务器还要检查云平台的安全组是否放行了对应端口。TCP长连接反复断开可能是中间网络设备空闲超时它会把长时间无数据的空闲连接当成死连接直接回收。这种场景下应用层应该有心跳机制比如每30秒发一个空包或者自定义心跳帧。另外防止半开连接可以给你的TCP socket设置settimeout并定期检测读写。5.4 排查模式不对、数据不流通的“绝招”如果数据完全不通我的一贯思路是分段隔离验证先用串口调试助手直接和串口设备通讯确认串口链路本身是好的、参数是对的。再起一个普通的TCP服务端/客户端程序用网络调试助手互发数据确认网络链路是好的。最后再启动转发工具把两边对接并打印完整收发日志。通过日志看数据到底卡在哪个环节。这套排查方法看着老套但真的比对着代码猜高效得多。因为很多时候问题根本不在转发工具本身而是串口设备的RX/TX接反了、USB转串口线坏了或者网线和防火墙出了问题。模块化排查能帮你用最小代价快速定位故障层。分享一个我自己的习惯不管这个转发工具最后给谁用我都会在代码里预留一个“debug模式”把每一帧从串口读到的原始hex和从TCP读到的原始hex都完整打印出来。平时不觉得有什么一旦设备端出了奇怪的问题这些原始日志就是救命稻草。还有就是在写这个工具的时候千万别嫌断线重连和日志记录麻烦这两个功能看着不起眼但在远程维护和无人值守场景下真的是决定工具可不可用的胜负手。你可以先跑通一个最简版本再加自动重连、加日志、加多客户端迭代着来这样整个过程会顺利很多。
网站建设高端定制企业官网