socketio与apscheduler并用

2019年4月28日 10:16

 

 说明

 
flask项目引入了flask-socketio提供websocket通信，同时需要flask-apscheduler完成定时任务。
 
 

 问题描述

 
项目初期只有socketio，并且以gunicron运行
 
gunicorn --worker-class eventlet -w 1 zhima_chat:app -b 0.0.0.0:5000 --access-logfile -
 
后来要引入apscheduler
 
以上面的方式运行，出现了问题。该如何将socketio与apschedeuler结合呢？
 
 

Tags: flask apscheduler python socketio
评论(134) 阅读(38423)

apscheduler如何传递参数给job

2015年7月03日 10:17 

import tornado
from apscheduler.schedulers.tornado import TornadoScheduler
sched = TornadoScheduler()


def job1(a, b, c):
    print "job1:", a,b,c


def job2(a, b, c):
    print "job2:", a,b,c

sched.add_job(job1, 'interval', seconds=1, args=["a", "b", "c"])
sched.add_job(job2, 'interval', seconds=1, kwargs={"a": "a", "b": "b", "c": "c"})
sched.start()

tornado.ioloop.IOLoop.instance().start()

Tags: apscheduler
评论(3) 阅读(9230)

apscheduler定时任务

2015年6月30日 16:50

使用apscheduler定时任务，可以使用interval任务+cron任务，interval定时更新cron配置信息，cron则实现作业计划。

通常用法

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()

def my_job():
    print 'hello world'

# 使用修饰器
@sched.scheduled_job('cron', id='my_job_id', second=10)
def hello():
    print "hello decorate"

# 轮循
sched.add_job(my_job, 'interval', seconds=5)
# 定时计划
sched.add_job(my_job, 'cron', second=5, minute=1, hour=12, day_of_week=2)
sched.start()

在tarnado中用法

import tornado
from apscheduler.schedulers.tornado import TornadoScheduler
sched = TornadoScheduler()

def my_job():
    print sched.get_jobs()

sched.add_job(my_job, 'interval', seconds=5, id="1")
sched.start()

tornado.ioloop.IOLoop.instance().start()

任务触发器比较

很多情况下，任务是根据数据库调整触发时间，时间改变了，如何判断触发器是否变化？
生成新trigger然后，专程字符串比较比较
# 触发器比较
str(job.trigger) != str(trigger)
# 修改触发器
sched.reschedule_job(job.id, trigger=trigger)
 

Tags: apscheduler
评论(3) 阅读(8574)