我在Python pyqt5中调用schedule时候遇到Bug，两个按钮执行两个线程，但在每个线程到达时间周期执行时会执行再次。这两个任务schedule设定的时间也明明不一样啊
import time
import schedule
from start_stop_demo import *
from PyQt5.QtWidgets import QApplication, QMainWindow
import sys
from PyQt5.QtCore import QThread
class MyThread(QThread):
def __init__(self):
super().__init__()
def run(self):
print("111")
send1()
class MyThread2(QThread):
def __init__(self):
super().__init__()
def run(self):
print("222")
send2()
class MyClass(QMainWindow, Ui_Form):
def __init__(self, parent=None):
super(MyClass, self).__init__(parent)
self.setupUi(self)
self.start_1.clicked.connect(self.start_work1)
self.stop_1.clicked.connect(self.end_work1)
self.start_2.clicked.connect(self.start_work2)
self.stop_2.clicked.connect(self.end_work2)
def start_work1(self):
self.my_thread = MyThread() # 创建线程
self.my_thread.start() # 开始线程
def end_work1(self):
if self.my_thread:
self.my_thread.terminate()
self.my_thread = None
def start_work2(self):
self.my_thread2 = MyThread2() # 创建线程
self.my_thread2.start() # 开始线程
def end_work2(self):
if self.my_thread2:
self.my_thread2.terminate()
self.my_thread2 = None
def send1():
# schedule.every(10).seconds.do(job1)
timeSet = "09:44" # format like "08:05"
schedule.every().day.at(timeSet).do(job1)
while True:
schedule.run_pending()
time.sleep(1)
def job1():
print("I'm working... in job1 start")
time.sleep(1)
print("I'm working... in job1 end")
def send2():
# schedule.every(10).seconds.do(job2)
timeSet = "09:29" # format like "08:05"
schedule.every().day.at(timeSet).do(job2)
while True:
schedule.run_pending()
time.sleep(1)
def job2():
print("I'm working... in job2 ")
if __name__ == '__main__':
# 添加以下代码可以防止窗口变小的情况
QtCore.QCoreApplication.setAttribute(QtCore.Qt.AA_EnableHighDpiScaling)
app = QApplication(sys.argv)
myWin = MyClass()
myWin.setWindowTitle("数据库与PLC通讯程序")
myWin.show()
sys.exit(app.exec_())
运行结果为：
111
222
I'm working in job1 start
I'm working in job1 start
I'm working in job1 end
I'm working in job1 end
我建议使用一个单独的线程来运行 schedule，而不是在每个点击按钮的线程中运行。这样可以确保 schedule 使用单独的线程，并且不会受到其他线程的干扰。
同时，为了避免任务重复执行多次，可以使用 schedule.every().minutes.do() 方法来代替 schedule.every().seconds.do() 方法，并将 minutes 参数设置得足够大，比如 5 分钟执行一次。
以下是 Python pyqt5 的代码片段：
import threading
import functools
import time
import schedule
from PyQt5.QtWidgets import QApplication, QPushButton, QMainWindow
class MainWindow(QMainWindow):
def __init__(self):
super().__init__()
self.initUI()
def initUI(self):
self.setGeometry(100, 100, 300, 200)
self.setWindowTitle('Schedule')
btn1 = QPushButton('Task 1', self)
btn1.resize(btn1.sizeHint())
btn1.move(50, 50)
btn1.clicked.connect(self.task1)
btn2 = QPushButton('Task 2', self)
btn2.resize(btn2.sizeHint())
btn2.move(150, 50)
btn2.clicked.connect(self.task2)
def task1(self):
print('Task 1')
def task2(self):
print('Task 2')
def run_schedule():
@functools.wraps(my_job)
def wrapper(*args, **kwargs):
try:
return my_job(*args, **kwargs)
except:
import traceback
print(traceback.format_exc())
return schedule.CancelJob
@with_logging
@catch_exceptions(cancel_on_failure=True)
def my_job():
print('Foo')
schedule.every(5).minutes.do(my_job) # 每 5 分钟执行一次
while True:
schedule.run_pending()
time.sleep(1)
if __name__ == '__main__':
app = QApplication([])
window = MainWindow()
window.show()
schedule_thread = threading.Thread(target=run_schedule)
schedule_thread.start() # 启动 schedule 的线程
app.exec_()
在这个代码片段中，我们首先创建了一个包含两个按钮的 PyQt5 窗口。当点击按钮时，它们会分别调用 task1 和 task2 方法。
然后我们创建了一个单独的函数 run_schedule，其中包含了你提供的 with_logging 和 catch_exceptions 装饰器，在每个任务执行前会打印出日志并捕获任何异常并进行处理。
在 run_schedule 函数中，我们将 my_job 方法添加到 schedule 中，并设置每 5 分钟执行一次。这个函数还包括了一个 while True 循环，它会不断地运行 schedule.run_pending() 以检查是否有任务需要执行。
最后，在 if __name__ == '__main__' 语句中，我们在一个新的线程中运行 run_schedule 方法。这样可以确保 schedule 使用单独的线程，并且不会受到其他线程的干扰。同时，在主线程中运行 PyQt5 的事件循环以保持界面的响应。
希望这个解决方案能对你有所帮助！