FlaskAPScheduler使用教程

项目中需要用到定时器和循环执行。去网上搜了一下&＃xff0c;比较常见的有一下集中。运用Python线程执行轮询操作&＃xff0c;也有运用Linux系统的Cron,Celery的文章最多&＃xff0c;但是太麻烦。看看就知道&＃xff0c;Celery 需要一个发送和接受消息的传输者。RabbitMQ 和 Redis 中间人的消息传输支持所有特性&＃xff0c;但也提供大量其他实验性方案的支持&＃xff0c;包括用 SQLite 进行本地开发。需要用到队列&＃xff0c;对于这点需求简直就是大材小用。最后找到了比较合适的Flask-APScheduler。

介绍

看看 github的flask-apscheduler介绍。

Loads scheduler configuration from Flask configuration.(支持从Flask中加载调度)
Loads job definitions from Flask configuration.(支持从Flask中加载任务配置)
Allows to specify the hostname which the scheduler will run on.(允许指定服务器运行任务)
Provides a REST API to manage the scheduled jobs.(提供Rest接口管理任务)
Provides authentication for the REST API.(提供Rest接口认证)

安装及配置

pip install Flask-APScheduler

在Flask配置文件中添加

SCHEDULER_API_ENABLED &＃61; True JOBS &＃61; [{&＃39;id&＃39;: &＃39;job_1h_data&＃39;,&＃39;func&＃39;: job_1h_data,&＃39;args&＃39;: &＃39;&＃39;,&＃39;trigger&＃39;: {&＃39;type&＃39;: &＃39;cron&＃39;,&＃39;day_of_week&＃39;: "0-6",&＃39;hour&＃39;: &＃39;*&＃39;,&＃39;minute&＃39;: &＃39;1&＃39;,&＃39;second&＃39;: &＃39;0&＃39;}},{&＃39;id&＃39;: &＃39;job_announce&＃39;,&＃39;func&＃39;: exchange_an,&＃39;args&＃39;: &＃39;&＃39;,&＃39;trigger&＃39;: &＃39;interval&＃39;,&＃39;seconds&＃39;: 300} ]

上面指定了每一小时获取所有货币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 &＃61; time.time()start_date &＃61; time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(current_local - 300))end_date &＃61; time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(current_local))announce &＃61; pro.query(&＃39;exchange_ann&＃39;, start_date&＃61;start_date, end_date&＃61;end_date)print(&＃39;请求交易所公告...&＃39;)for x in announce.values:s &＃61; {&＃39;title&＃39;: x[0],&＃39;content&＃39;: x[1],&＃39;type&＃39;: x[2],&＃39;url&＃39;: x[3],&＃39;datetime&＃39;: x[4]}value &＃61; json.dumps(s)print(value)mqttClient.publish(&＃39;system/ex_announce&＃39;, value)

动态添加任务

# coding:utf-8 from apscheduler.schedulers.blocking import BlockingScheduler import datetimedef aps_test(x):print datetime.datetime.now().strftime(&＃39;%Y-%m-%d %H:%M:%S&＃39;), xscheduler &＃61; BlockingScheduler() scheduler.add_job(func&＃61;aps_test, args&＃61;(&＃39;定时任务&＃39;,), trigger&＃61;&＃39;cron&＃39;, second&＃61;&＃39;*/5&＃39;) scheduler.add_job(func&＃61;aps_test, args&＃61;(&＃39;一次性任务&＃39;,), next_run_time&＃61;datetime.datetime.now() &＃43; datetime.timedelta(seconds&＃61;12)) scheduler.add_job(func&＃61;aps_test, args&＃61;(&＃39;循环任务&＃39;,), trigger&＃61;&＃39;interval&＃39;, seconds&＃61;3, id&＃61;&＃39;interval_task&＃39;)scheduler.start() """ 暂停任务 """ scheduler.pause_job(&＃39;interval_task&＃39;) """ 恢复任务 """ scheduler.resume_job(&＃39;interval_task&＃39;) """ 删除任务 """ scheduler.remove_job(&＃39;interval_task&＃39;)