APScheduler是一個強大的Python調度庫,它支持多種后端存儲方式。MySQL是其中之一。下面我們來介紹如何使用APScheduler和MySQL來實現任務的調度。
首先,我們需要安裝APScheduler和MySQL的Python庫:
pip install apscheduler pip install mysql-connector-python
其次,我們需要建立MySQL數據庫,并創建一個任務表來存儲任務信息:
CREATE TABLE tasks ( id INT AUTO_INCREMENT PRIMARY KEY, job_id VARCHAR(255), func VARCHAR(255), args VARCHAR(255), kwargs VARCHAR(255), trigger VARCHAR(255), next_run_time DATETIME );
然后,我們需要在Python中連接到數據庫:
import mysql.connector conn = mysql.connector.connect( host="localhost", user="root", password="password", database="test" )
接著,我們創建一個MySQLJobStore實例,并將其作為參數傳遞給APScheduler的Scheduler類:
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.mysql import MySQLJobStore jobstore = MySQLJobStore( username='root', password='password', database='test', host='localhost', port=3306 ) scheduler = BackgroundScheduler(jobstores=jobstore)
現在,我們可以定義一個任務函數,并將其添加到調度器中:
def job_func(): print("Hello, world!") scheduler.add_job( job_func, 'interval', seconds=10, id='my_job_id' )
最后,我們需要啟動調度器并運行任務:
scheduler.start() try: while True: pass except (KeyboardInterrupt, SystemExit): scheduler.shutdown()
現在,每隔10秒鐘,我們都會看到一條“Hello, world!”的輸出。此外,我們的任務信息也會被保存在MySQL數據庫中。