同一QThread中多QTimer任务串行执行的原因及优化方案问询

问题描述 #

在单个QThread中运行多个QTimer，期望每个定时器按设定间隔独立触发，触发后通过QSemaphore竞争公共资源：先获取资源的任务执行，后续任务阻塞等待；已完成的任务到下一次触发时间时，立即进入资源等待队列。但实际所有任务串行执行，下一个任务需等前一个完成才启动，且第二轮任务的触发时间为上一轮最后一个任务结束时间加超时时间。

尝试过将不同定时器放入不同子线程实现并行，但无法保证任务公平访问资源，需要找到两全的方法。

现有代码 #

from PyQt5.QtCore import QObject, QThread, QTimer, QSemaphore, pyqtSignal
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QPushButton
import time
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S")

class Worker(QObject):
    stopped = pyqtSignal()
    stop_signal = pyqtSignal()

    def __init__(self):
        super().__init__()
        self.semaphore = QSemaphore(1)
        self.runners = []
        self.threads = []

    def startTask(self):
        self.runners = []
        for i in range(1, 5):
            timer = Timer(f"runner_{i}", self)
            self.runners.append(timer)

        for timer in self.runners:
            timer.start_task()

    def stopTask(self):
        print("client stopTask is called.")
        for timer in self.runners:
            timer.stop_task()
        self.stopped.emit()


class Timer(QTimer):
    def __init__(self, name, client):
        super().__init__()
        self.name = name
        self.client = client
        self.timeout.connect(self.timer_timeout)

    def timer_timeout(self):
        logger.info(f"{self.name} 启动。")
        self.client.semaphore.acquire(1)
        time.sleep(1)
        logger.info(f"{self.name} 停止。")
        self.client.semaphore.release(1)

    def start_task(self):
        self.start(4000)

    def stop_task(self):
        self.stop()

class Widget(QWidget):
    stop_signal = pyqtSignal()

    def __init__(self):
        super().__init__()
        self._layout = QVBoxLayout(self)
        self.start_btn = QPushButton("start")
        self.stop_btn = QPushButton("stop")
        self._layout.addWidget(self.start_btn)
        self._layout.addWidget(self.stop_btn)
        self.start_btn.clicked.connect(self.start)
        self.stop_btn.clicked.connect(self.stop)

    def start(self):
        self._thread = QThread()
        self.worker = Worker()
        self.worker.moveToThread(self._thread)
        self._thread.started.connect(self.worker.startTask)

        self.stop_signal.connect(self.worker.stopTask)
        self.worker.stopped.connect(self._thread.quit)
        self.worker.stopped.connect(self.worker.deleteLater)

        self._thread.finished.connect(self._thread.deleteLater)
        self._thread.start()

    def stop(self):
        self.stop_signal.emit()

if __name__ == "__main__":
    import sys
    from PyQt5.QtWidgets import QApplication
    app = QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())

当前运行结果 #

2023-09-18 23:03:11: runner_4 启动。
2023-09-18 23:03:12: runner_4 停止。
2023-09-18 23:03:12: runner_3 启动。
2023-09-18 23:03:13: runner_3 停止。
2023-09-18 23:03:13: runner_2 启动。
2023-09-18 23:03:14: runner_2 停止。
2023-09-18 23:03:14: runner_1 启动。
2023-09-18 23:03:15: runner_1 停止。
2023-09-18 23:03:19: runner_4 启动。
2023-09-18 23:03:20: runner_4 停止。
2023-09-18 23:03:20: runner_3 启动。
2023-09-18 23:03:21: runner_3 停止。

问题原因 #

1. QThread事件循环串行处理：同一QThread的事件循环是单线程串行执行所有槽函数的，即使四个QTimer同时触发timeout信号，这些信号的槽函数也会被排队依次执行。
2. 阻塞线程导致定时器计时中断：time.sleep(1)会阻塞当前QThread，事件循环无法处理定时器的计时逻辑，导致定时器的下一次触发时间从槽函数执行完毕后才开始计算，而非设定的固定间隔。

解决方案：定时器触发+线程池执行+公平同步 #

将定时器的触发逻辑与任务执行逻辑分离：定时器仅负责按时触发任务通知，实际的资源占用任务交由Qt线程池执行；同时使用公平模式的QSemaphore（或QMutex）保证任务公平竞争资源。

修改后的代码 #

from PyQt5.QtCore import QObject, QThread, QTimer, QSemaphore, pyqtSignal, QRunnable, QThreadPool
from PyQt5.QtWidgets import QWidget, QVBoxLayout, QPushButton
import time
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S")

class TaskRunnable(QRunnable):
    def __init__(self, name, semaphore):
        super().__init__()
        self.name = name
        self.semaphore = semaphore

    def run(self):
        logger.info(f"{self.name} 进入资源等待队列。")
        # 公平模式获取信号量，保证先触发的任务先获取资源
        self.semaphore.acquire(1)
        try:
            logger.info(f"{self.name} 开始执行任务。")
            time.sleep(1)
            logger.info(f"{self.name} 任务执行完毕，释放资源。")
        finally:
            self.semaphore.release(1)

class Worker(QObject):
    stopped = pyqtSignal()
    stop_signal = pyqtSignal()

    def __init__(self):
        super().__init__()
        # 创建公平模式的信号量，保证等待队列先进先出（Qt 5.15+支持）
        self.semaphore = QSemaphore(1)
        self.semaphore.setFair(True)
        self.runners = []
        self.thread_pool = QThreadPool.globalInstance()

    def startTask(self):
        self.runners = []
        for i in range(1, 5):
            timer = Timer(f"runner_{i}", self)
            self.runners.append(timer)

        for timer in self.runners:
            timer.start_task()

    def submit_task(self, name):
        runnable = TaskRunnable(name, self.semaphore)
        self.thread_pool.start(runnable)

    def stopTask(self):
        print("client stopTask is called.")
        for timer in self.runners:
            timer.stop_task()
        # 等待线程池所有任务完成后再停止
        self.thread_pool.waitForDone()
        self.stopped.emit()


class Timer(QTimer):
    def __init__(self, name, client):
        super().__init__()
        self.name = name
        self.client = client
        self.timeout.connect(self.timer_timeout)

    def timer_timeout(self):
        logger.info(f"{self.name} 定时器触发。")
        # 提交任务到线程池，不阻塞定时器所在线程
        self.client.submit_task(self.name)

    def start_task(self):
        self.start(4000)

    def stop_task(self):
        self.stop()

class Widget(QWidget):
    stop_signal = pyqtSignal()

    def __init__(self):
        super().__init__()
        self._layout = QVBoxLayout(self)
        self.start_btn = QPushButton("start")
        self.stop_btn = QPushButton("stop")
        self._layout.addWidget(self.start_btn)
        self._layout.addWidget(self.stop_btn)
        self.start_btn.clicked.connect(self.start)
        self.stop_btn.clicked.connect(self.stop)

    def start(self):
        self._thread = QThread()
        self.worker = Worker()
        self.worker.moveToThread(self._thread)
        self._thread.started.connect(self.worker.startTask)

        self.stop_signal.connect(self.worker.stopTask)
        self.worker.stopped.connect(self._thread.quit)
        self.worker.stopped.connect(self.worker.deleteLater)

        self._thread.finished.connect(self._thread.deleteLater)
        self._thread.start()

    def stop(self):
        self.stop_signal.emit()

if __name__ == "__main__":
    import sys
    from PyQt5.QtWidgets import QApplication
    app = QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())

方案说明 #

1. 定时器与任务分离：QTimer仅负责按时触发timeout信号，槽函数中只是将任务提交到线程池，不会阻塞定时器所在线程，因此定时器能严格按照设定的4秒间隔触发。
2. 线程池执行任务：任务的资源占用逻辑放到QRunnable中，由Qt全局线程池执行，实现任务的并行等待与执行。
3. 公平资源访问：使用公平模式的QSemaphore（Qt 5.15+支持），保证先进入等待队列的任务优先获取资源，实现公平竞争。若使用旧版本Qt，可替换为QMutex(QMutex.Fair)来实现公平锁。

预期运行效果 #

• 第一轮所有定时器会在同一时间触发（如23:03:11），并提交任务到线程池。
• 第一个任务获取资源执行，其余三个进入等待队列。
• 任务完成后立即释放资源，等待队列中的下一个任务开始执行。
• 每个定时器的下一次触发严格按照4秒间隔（如runner_4的下一次触发时间为23:03:15），触发后立即进入资源等待队列，无需等待上一轮所有任务完成。

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