Flask APScheduler使用教程

專案中需要用到定時器和迴圈執行。去網上搜了一下，比較常見的有一下集中。運用python執行緒執行輪詢操作，也有運用linux系統的cron,celery的文章最多，但是太麻煩。看看就知道，celery 需要乙個傳送和接受訊息的傳輸者。rabbitmq 和 redis 中間人的訊息傳輸支援所有特性，但也提供大量其他實驗性方案的支援，包括用 sqlite 進行本地開發。需要用到佇列，對於這點需求簡直就是大材小用。最後找到了比較合適的flask-apscheduler。

看看 github的flask-apscheduler介紹。

pip install flask-apscheduler

在flask配置檔案中新增

scheduler_api_enabled =
true
jobs =[}
,]

上面指定了每一小時獲取所有貨幣24h最高位以及交易所公告。

def
exchange_an()
:"""
:param start_date: 開始時間 yyyy-mm-dd hh:mm:ss
:param end_date: 結束時間 yyyy-mm-dd hh:mm:ss
:return: 推送訊息,保持資料庫
"""current_local = time.time(
) start_date = time.strftime(
"%y-%m-%d %h:%m:%s"
, time.localtime(current_local -
300)
) end_date = time.strftime(
"%y-%m-%d %h:%m:%s"
, time.localtime(current_local)
) announce = pro.query(
'exchange_ann'
, start_date=start_date, end_date=end_date)
print
('請求交易所公告...'
)for x in announce.values:
s =value = json.dumps(s)
print
(value)
mqttclient.publish(
'system/ex_announce'
, value)

# coding:utf-8
from apscheduler.schedulers.blocking import blockingscheduler
import datetime
defaps_test
(x):
print datetime.datetime.now(
).strftime(
'%y-%m-%d %h:%m:%s'
), x
scheduler = blockingscheduler(
)scheduler.add_job(func=aps_test, args=
('定時任務',)
, trigger=
'cron'
, second=
'*/5'
)scheduler.add_job(func=aps_test, args=
('一次性任務',)
, next_run_time=datetime.datetime.now(
)+ datetime.timedelta(seconds=12)
)scheduler.add_job(func=aps_test, args=
('迴圈任務',)
, trigger=
'interval'
, seconds=3,
id='interval_task'
)scheduler.start(
)"""
暫停任務
"""scheduler.pause_job(
'interval_task'
)"""
恢復任務
"""scheduler.resume_job(
'interval_task'
)"""
刪除任務
"""scheduler.remove_job(
'interval_task'
)

apscheduler支援新增三種方式的任務，分別是定時任務，一次性任務及迴圈任務。同時也包含了對任務的控制。

因為是單機版本，所以指定伺服器執行任務，rest介面管理任務，rest介面認證就不寫了。後續有需求在繼續。

Flask APScheduler使用教程

Flask APScheduler定時排程任務

flask apscheduler 定時任務框架

koa generic session 使用教程

Flask APScheduler使用教程

Flask APScheduler定時排程任務

flask apscheduler 定時任務框架

koa generic session 使用教程

相關推薦