Python并行处理与/dev/ttyS0端口共享问题求助

问题根源 #

1. 串口资源冲突：原代码中两个函数各自创建GSM实例，同时打开同一个串口/dev/ttyS0，但串口是独占式资源，无法被多个进程/实例同时访问，直接导致端口访问错误。
2. 死锁原因：之前用锁包裹整个无限循环，导致持有锁的线程在等待输入（input()阻塞）时无法释放锁，另一个线程永远拿不到锁，引发死锁；锁的粒度设计错误，没有针对单次串口读写操作做保护。
3. 异步方案失效：serial库的读写操作是同步阻塞实现，没有原生异步支持，强行用异步框架无法解决资源竞争问题。

解决方案：共享实例+线程锁保护 #

采用多线程（IO密集型任务最优选择），让两个函数共享同一个GSM实例，并用互斥锁保护所有串口读写操作，确保同一时间只有一个线程访问串口。

修改后完整代码 #

import serial
from time import sleep
from gotify import Gotify
import threading

ctrlZ = chr(26)

gsm = {
        'check':'AT',
        'sig_quality':'AT+CSQ',
        'ready':'AT+CPIN?',
        #tech details
        'manufacturer':'AT+CGMI',
        'model':'AT+CGMM',
        'imsi':'AT+CIMI',
        'iccid':'AT+CCID',
        'net_reg_stat':'AT+CREG?',
        'scan_networks':'AT+COPS=?',
        'apn':'AT+CGDCONT?',
        'operator':'AT+COPS?',
        #tackling with GSM
        'check_sms':'AT+CMGF=1',
        'list_sms':'AT+CMGL="ALL"',
        'dial_last':'ATDL',
        'hangup':'ATH',
        'find_number':'AT+CNUM',
        'answer':'ATA',
    }

for key, value in gsm.items():
    gsm[key] = value + '\n'


class GSM:
    def __init__(self, port='/dev/ttyS0', baudrate=115200):
        self.ser = serial.Serial(port, baudrate)
        self.lock = threading.Lock()  # 添加互斥锁
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.ser.close()

    def transceive(self, message):
        with self.lock:  # 用锁保护串口写+读操作
            self.ser.write(message.encode())
            self.ser.read_until()

    def receive(self):
        with self.lock:  # 用锁保护串口读操作
            return self.ser.read_until().decode()[:-2]

    def auto_answer(self, accept):
        self.transceive('ATS0=' + accept + '\n')

    def dial(self, number):
        self.transceive('ATD' + number + ';\n')

    def send_sms(self, number, text):
        self.transceive('AT+CMGS=' + '"' + number + '"' + '\n' + text + ctrlZ)

    def unlock_enable(self, status):
        if status:
            val = '1'
        else:
            val = '0'
        self.transceive('AT+CFUN=' + val + '\n')
        while True:
            answer = self.receive()
            if answer == '+CPIN: SIM PIN':
                print('Modem powered!')
            elif answer == 'OK':
                break

    def unlock_sim(self, pin):
        self.transceive('AT+CPIN=' + pin + '\n')
        while True:
            response = self.receive()
            if response == 'ERROR':
                raise Exception('Incorrect PIN!')
            elif response == 'OK':
                print('SIM activated!')
                break

    def wait(self, delay):
        self.transceive('WAIT=' + delay + '\n')


def init(g, pin):
    g.transceive(gsm['ready'])
    if g.receive() != '+CPIN: READY':
        g.unlock_enable(True)
        g.unlock_sim(pin)
    sleep(3)
    g.transceive(gsm['check_sms'])
    while True:
        response = g.receive()
        if response == 'ERROR':
            print('Cannot send SMS!')
            break
        elif response == 'OK':
            print('Ready for sending SMS!')
            break


def serial_spit(g, pin, initial_number):
    number = initial_number

    while True:    
        cmd = input(f'''Introduce the command:
                    c - change the phone number (now it is {number} )
                    d - dial
                    h - hangup
                    m - message\n\n''')
        if cmd == 'c':
            number = input('Please, introduce the phone number you would like to use from now on.\n')
        elif cmd == 'd':
            g.dial(number)
        elif cmd == 'h':
            g.transceive(gsm['hangup'])
        elif cmd == 'm':
            txt = input('Please, introduce the next message.\n')
            if txt == '' :
                continue
            g.send_sms(number, txt)
        else:
            print('Command not recognized')
    

def serial_digest(g):
    last_dialing_number = 0

    gotify = Gotify(
            base_url='http://localhost:80',
            app_token='A_8APqsvE8C5Rom',
    )
    
    while True:
        call_status = ''
        data = g.receive()

        if data == 'RING':
            call_status='Ring'
            g.transceive('AT+CLCC\r\n')
            data = g.receive()
            if data.startswith('+CLCC'):
                for x in data[len('+CLCC: '):-len('\r\n')].split(','):
                    if x.startswith('"') and x.endswith('"'):
                        last_dialing_number = x.strip('"')
                gotify.create_message(
                    message=call_status,
                    title=last_dialing_number,
                )
                print(last_dialing_number)
                print(call_status)
                print()
            elif data == 'NO CARRIER':
                call_status='Hang'
                gotify.create_message(
                    message=call_status,
                    title=last_dialing_number,
                )
                print(last_dialing_number)
                print(call_status)
                print()
            elif data.startswith('+CMT'):
                [number,_,date,hour] = [x.strip('"') for x in data[len('+CMT: '):-len('\r\n')].split(',')]
                message = g.receive()
                message += '\n' + date + ' ' + hour
                title = number[2:]
                print(title)
                print(message)
                print()
                gotify.create_message(
                    message=message,
                    title=title,
                )

if __name__ == "__main__":
    pin = '0000'
    initial_number = '0744604883'

    # 主线程初始化GSM实例，避免重复打开串口
    with GSM() as g:
        cmd = input('If you would like to set the PIN press nothing but the return key.\n')
        if cmd == '':
            pin = input('Please, introduce the PIN.\n')
        init(g, pin)

        # 创建并启动两个线程
        spit_thread = threading.Thread(target=serial_spit, args=(g, pin, initial_number))
        digest_thread = threading.Thread(target=serial_digest, args=(g,))
        
        spit_thread.daemon = True  # 主线程退出时自动终止子线程
        digest_thread.daemon = True
        
        spit_thread.start()
        digest_thread.start()

        # 主线程等待子线程结束（这里因为是无限循环，实际会一直运行直到手动终止）
        spit_thread.join()
        digest_thread.join()

关键改动说明 #

1. 添加线程锁：在GSM类中新增threading.Lock()，并在transceive、receive等所有涉及串口读写的方法中用with self.lock:包裹操作，确保同一时间只有一个线程访问串口，避免资源竞争。
2. 共享GSM实例：主线程中创建唯一的GSM实例，完成初始化后传递给两个线程，避免重复打开同一个串口。
3. 重构函数参数：修改serial_spit和serial_digest，让它们接收共享的GSM实例作为参数，不再各自创建实例。
4. 线程守护设置：将子线程设为守护线程，这样当主线程意外退出时，子线程会自动终止，避免残留进程。
5. 锁粒度控制：只对单次串口读写操作加锁，而非整个循环。serial_spit中等待input()时不会持有锁，serial_digest可以正常读取串口数据，避免死锁。

内容的提问来源于stack exchange，提问作者pauk