Python3 实现串口同时读写：多线程日志记录与退出机制

小石头

发布时间：2026-10-12| 3人浏览过

原创

本文介绍如何使用 Python3 的 threading 模块实现串口的同时读写功能。通过创建独立的接收和发送线程，程序能够实时记录收发数据到日志文件，并在接收到特定字符（如 'q'）时安全退出。代码示例展示了串口初始化、线程管理及异常处理的具体实现。

实现思路与核心逻辑

在串口通信应用中，通常需要同时进行数据的发送与接收。Python 提供了 threading 模块，允许我们创建多个线程来并行处理这些任务。本方案的核心在于：

使用两个独立的线程分别负责读取和发送数据。
将发送和接收的数据分别记录到不同的日志文件中，便于后续分析。
通过检测特定字符（如 'q'）来触发程序的正常退出。

代码实现详解

以下代码展示了如何实现上述功能。代码中定义了一个 SerThread 类，封装了串口的初始化和线程管理逻辑。

import threading,time
import serial
import string
 
 
class SerThread:
  def __init__(self, Port=0):
    #初始化串口、blog文件名称
    self.my_serial = serial.Serial()
    self.my_serial.port=Port
    self.my_serial.baudrate = 9600
    self.my_serial.timeout = 1    
    self.alive = False
    self.waitEnd = None
    fname=time.strftime("%Y%m%d")#blog名称为当前时间
    self.rfname='r'+fname #接收blog名称
    self.sfname='s'+fname #发送blog名称
    self.thread_read= None
    self.thread_send=None   
       
  def waiting(self):
    # 等待event停止标志
    if not self.waitEnd is None:
      self.waitEnd.wait()
  
  def start(self):
    #开串口以及blog文件 
    self.rfile=open(self.rfname,'w')
    self.sfile=open(self.sfname,'w')
    self.my_serial.open()
       
    if self.my_serial.isOpen():
      self.waitEnd = threading.Event()
      self.alive = True
      
      self.thread_read = threading.Thread(target=self.Reader)
      self.thread_read.setDaemon(True)
      
      self.thread_send=threading.Thread(target=self.Sender)
      self.thread_send.setDaemon(True)
      
      self.thread_read.start()
      self.thread_send.start()
      return True
    else:
      return False
  
  def Reader(self):
    while self.alive:
      try:
        n=self.my_serial.inWaiting()
        data=''
        if n:
          data= self.my_serial.read(n).decode('utf-8')       
          print ('recv'+' '+time.strftime("%Y-%m-%d %X")+' '+data.strip())
          print (time.strftime("%Y-%m-%d %X:")+data.strip(),file=self.rfile)
          if len(data)==1 and ord(data[len(data)-1])==113: #收到字母q，程序退出
            break
      except Exception as ex:
        print (ex)
        
    self.waitEnd.set()
    self.alive = False
  
  def Sender(self):
    while self.alive:
      try:
        snddata=input("input data:\n")
        self.my_serial.write(snddata.encode('utf-8'))
        print ('sent'+' '+ time.strftime("%Y-%m-%d %X"))
        print (snddata,file=self.sfile) 
        
      except Exception as ex:
        print (ex)
    
    self.waitEnd.set()
    self.alive = False          
        
  def stop(self):
    self.alive = False
    #self.thread_read.join()
    #self.thread_send.join()
    if self.my_serial.isOpen():
      self.my_serial.close()
    self.rfile.close()
    self.sfile.close()
      
if __name__ == '__main__':  
  
  ser = SerThread('com4')
  try:
    if ser.start():
      ser.waiting()
      ser.stop()
    else:
      pass;      
  except Exception as ex:
    print (ex)
  
  if ser.alive:
    ser.stop()
  
  print ('End OK .');
  del ser;