计划和计时器的 PYQT5 线程问题

Question

我正在使用 PYQT5 构建一个 GUI，我正在使用 APScheduler 来管理我想要运行的作业。我将调度程序项和计时器项分解成自己的类，然后将它们连接到主文件中。

我遇到的问题是，一旦计时器完成一个周期，我尝试将时间添加到 Timer 类并重新开始它以进行下一次倒计时，然后调度程序应该再次运行。我收到两个错误或警告，我不知道如何解决它们。它们是：

QObject::killTimer: Timers cannot be stopped from another thread
QObject::startTimer: Timers cannot be started from another thread

一旦这些被抛出，GUI 更新但不再倒计时。我将附上我发现会重现错误的最简单版本。非常感谢您的任何帮助，并感谢您的宝贵时间。

Main.py

import sys
from PyQt5 import QtCore
from PyQt5 import QtWidgets

from Timer import Timer
from Schedule import Scheduler

import datetime


class MyMainWindow(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
 
        central_widget = QtWidgets.QWidget()
        self.setCentralWidget(central_widget)
        vbox = QtWidgets.QVBoxLayout()
        central_widget.setLayout(vbox)

        self.start_pushButton = QtWidgets.QPushButton()
        self.start_pushButton.setText("Start")
        self.start_pushButton.clicked.connect(self.start_schedule)
        vbox.addWidget(self.start_pushButton)

        self.pages_qsw = QtWidgets.QStackedWidget()
        vbox.addWidget(self.pages_qsw)
        self.time_passed_qll = QtWidgets.QLabel()
        vbox.addWidget(self.time_passed_qll)

        self.my_timer = Timer()
        self.my_timer.get_seconds.connect(self.update_gui)

        self.sch = Scheduler() 
 
    def start_schedule(self):
        self.sch.add(self.hello)
        self.sch.start()
        self.start_my_timer()


    def start_my_timer(self):
        next_run = self.sch.next_occurance().replace(tzinfo=None) # This removes the time zone.
    
        a = datetime.datetime.now()
        difference = next_run - a

        self.my_timer.addSecs(difference.seconds)
        self.my_timer.timer_start()

    def hello(self):
        print("hello world")
        self.start_my_timer()
    
    @QtCore.pyqtSlot(str)
    def update_gui(self,seconds):
        self.time_passed_qll.setText(str(seconds))

app = QtWidgets.QApplication(sys.argv)
main_window = MyMainWindow()
main_window.show()
sys.exit(app.exec_())

Timer.py

from PyQt5.QtCore import QTimer, pyqtSignal
from PyQt5.QtGui import *
from PyQt5.QtWidgets import *

import datetime

class Timer(QTimer):
    get_seconds = pyqtSignal(str)
    
    def __init__(self):
        super().__init__()
        self.time_left = 1
        self.timeout.connect(self.timer_timeout)

    def addSecs(self, secs):
        self.time_left += secs

    def timer_start(self):
        self.start(1000)
        self.update_gui()

    def timer_timeout(self):
        self.time_left -= 1

        if self.time_left <= 0:
            self.stop()

        self.update_gui()

    def update_gui(self):
        self.get_seconds.emit(str(self.time_left))

Schedule.py

from datetime import datetime
from apscheduler.schedulers.qt import QtScheduler   

class Scheduler():
    def __init__(self):
        self.id = 'test_job'
        self.sched = QtScheduler()

    def add(self,job_function):
        self.sched.add_job(job_function, 'cron', day_of_week='mon-fri',hour ='9-18',minute = '2,7,12,17,22,27,32,37,42,47,52,57',second = '5', id=self.id)

    def start(self):
        self.sched.start()

    def next_occurance(self):
        for job in self.sched.get_jobs():
            if job.id == self.id:
                return job.next_run_time  

Answer 1

正如错误所解释的，您无法从另一个线程启动和停止 QTimer，并且由于 APScheduler 在不同的线程上工作，因此您的问题的原因：self.hello 是从 APScheduler 线程调用的，而不是从您的 Timer 所在的线程已创建。

要访问在不同线程中创建的对象，您需要使用信号和槽，以便让 Qt 管理不同线程之间的通信。

因此，解决方案可能是通过从 QObject 继承来子类化您的调度程序（以便能够创建信号并连接到它们），然后在每次执行作业时使用自定义信号并使用 that 重启定时器的信号。

为此，我使用了createJob 函数，它实际运行作业并在作业开始时发出started 信号，并在完成时发出completed。

很遗憾，我无法测试以下代码，因为我现在无法安装 APScheduler，但逻辑应该没问题。

Main.py

class MyMainWindow(QtWidgets.QMainWindow):
    def __init__(self):
        # ...
        self.sch = Scheduler()
        self.sch.completed.connect(self.start_my_timer)

    def hello(self):
        print("hello world")
        # no call to self.start_my_timer here!

Schedule.py

from datetime import datetime
from apscheduler.schedulers.qt import QtScheduler
from PyQt5 import QtCore

class Scheduler(QtCore.QObject):
    started = QtCore.pyqtSignal(object)
    completed = QtCore.pyqtSignal(object)
    def __init__(self):
        self.id = 'test_job'
        self.sched = QtScheduler()

    def add(self, job_function, *args, **kwargs):
        self.sched.add_job(self.createJob(job_function), 'cron', 
            day_of_week='mon-fri', hour='9-18',
            minute='2,7,12,17,22,27,32,37,42,47,52,57',
            second='5', id=self.id, *args, **kwargs)

    def createJob(self, job_function):
        def func(*args, **kwargs):
            self.started.emit(job_function)
            job_function(*args, **kwargs)
            self.completed.emit(job_function)
        return func

    def start(self):
        self.sched.start()

    def next_occurance(self):
        for job in self.sched.get_jobs():
            if job.id == self.id:
                return job.next_run_time

请注意，我使用 job_function 参数（在您的情况下是对工作的引用，self.hello）发出 started 和 completed 信号，这可能有助于识别工作如果您想对多个工作做出不同的反应。我还添加了对位置和关键字参数的基本支持。

另外请注意，我只提供了一个非常基本的 实现（您的函数只打印一条消息）。如果您需要在作业函数中与 UI 元素交互，QTimer 也会出现同样的问题，因为不允许从主 Qt 线程之外的线程访问 UI 元素。

在这种情况下，您需要找到另一种方法。例如，您可以添加一个作业（实际上并非从调度程序运行）并以该作业作为参数发出一个信号，然后连接到将在主线程中实际运行该作业的函数。

Main.py

class MyMainWindow(QtWidgets.QMainWindow):
    def __init__(self):
        # ...
        self.sch = Scheduler()
        self.sch.startJob.connect(self.startJob)

    def startJob(self, job, args, kwargs):
        job(*args, **kwargs)
        self.start_my_timer()

    def hello(self):
        self.someLabel.setText("hello world")

Schedule.py

from datetime import datetime
from apscheduler.schedulers.qt import QtScheduler
from PyQt5 import QtCore

class Scheduler(QtCore.QObject):
    startJob = QtCore.pyqtSignal(object, object, object)
    # ...
    def add(self, job_function, *args, **kwargs):
        self.sched.add_job(self.createJob(job_function, args, kwargs), 'cron', 
            day_of_week='mon-fri', hour='9-18',
            minute='2,7,12,17,22,27,32,37,42,47,52,57',
            second='5', id=self.id)

    def createJob(self, job_function, args, kwargs):
        def func():
            self.starJob.emit(job_function, args, kwargs)
        return func

如前所述，上面的代码未经测试，您需要检查可能的错误（也许我在通配符参数上犯了一些错误）。

最后，一些小建议：

1. 实际上需要使用pyqtSlot装饰器的情况非常少见；有趣的是，使用它们通常会导致问题或意外行为。
2. 通常最好将信号参数保持原样而不进行任何转换，因此您不应该将时间转换为get_seconds 信号的字符串；此外，QLabel 可以接受使用 setNum() 的数值（对于浮点数和整数数）。
3. 对空格更加小心（我指的是self.sched.add_job）：对于关键字参数，空格只能在逗号​​之后存在（阅读更多Style Guide for Python Code）；虽然它实际上并不代表问题，但它极大地提高了可读性。