如何确保QThread中串行设备的两个读写任务不重叠？

问题本质 #

即使将读写操作放在同一线程，若串口设备的读写逻辑未做互斥保护，或信号连接方式不当，仍可能出现并发执行的情况（比如非阻塞串口的异步操作、跨线程信号触发的槽函数未正确排队）。以下是几种可靠的解决方法：

方案一：用QMutex实现读写互斥 #

通过互斥锁确保同一时间只有一个操作访问串口设备，是最直接的线程安全保障方式。

修改后的SerialWorker代码：

from PyQt6.QtCore import QObject, pyqtSignal, QMutex, QMutexLocker

class SerialWorker(QObject):
    updateUI = pyqtSignal()

    def __init__(self, device):
        super().__init__()
        self.device = device
        self._mutex = QMutex()  # 初始化互斥锁

    def read(self):
        # QMutexLocker自动管理锁的加解锁，避免异常导致锁未释放
        with QMutexLocker(self._mutex):
            self.device.read()
            self.updateUI.emit()

    def write(self):
        with QMutexLocker(self._mutex):
            self.device.write()

方案二：用任务队列串行化所有操作 #

如果需要更灵活的任务管理（比如支持任务优先级、取消任务），可以将读写任务封装后放入队列，让worker线程循环串行处理队列中的任务。

代码示例：

from PyQt6.QtCore import QObject, pyqtSignal, QMutex, QWaitCondition
from queue import Queue

class SerialWorker(QObject):
    updateUI = pyqtSignal()
    _task_queue = Queue()
    _mutex = QMutex()
    _condition = QWaitCondition()
    _running = True

    def __init__(self, device):
        super().__init__()
        self.device = device

    def start_processing(self):
        # 启动任务处理循环，需在worker线程中调用
        while self._running:
            self._mutex.lock()
            # 等待队列中有任务
            while self._task_queue.empty() and self._running:
                self._condition.wait(self._mutex)
            if not self._running:
                self._mutex.unlock()
                break
            task = self._task_queue.get()
            self._mutex.unlock()
            # 执行串行任务
            task()

    def stop_processing(self):
        # 停止任务循环
        self._mutex.lock()
        self._running = False
        self._condition.wakeAll()
        self._mutex.unlock()

    def add_read_task(self):
        # 提交读任务到队列
        def read_task():
            self.device.read()
            self.updateUI.emit()
        self._mutex.lock()
        self._task_queue.put(read_task)
        self._condition.wakeAll()
        self._mutex.unlock()

    def add_write_task(self):
        # 提交写任务到队列
        def write_task():
            self.device.write()
        self._mutex.lock()
        self._task_queue.put(write_task)
        self._condition.wakeAll()
        self._mutex.unlock()

使用时，需在worker线程中调用start_processing()，通过add_read_task()和add_write_task()提交任务，而非直接触发读写方法。

方案三：确保信号连接方式为QueuedConnection #

跨线程发送信号时，手动指定QueuedConnection，确保槽函数在worker线程的事件循环中排队执行（PyQt跨线程默认会自动使用该连接方式，但手动指定更稳妥）：

from PyQt6.QtCore import Qt

# 连接信号与槽时指定连接类型
other_thread_read_signal.connect(worker.read, Qt.ConnectionType.QueuedConnection)
other_thread_write_signal.connect(worker.write, Qt.ConnectionType.QueuedConnection)

注意：若串口设备的读写逻辑本身是异步的，仍需配合互斥锁使用该方案。

内容的提问来源于stack exchange，提问作者André Santos