apscheduler是Python中一个很流行的任务调度库。它可以用来实现定时任务,Cron任务,以及更复杂的任务调度方案。 主要特性有:
- 支持Cron表达式,定时任务,间隔任务等多种调度类型
- 支持一次性任务和重复任务
- 支持任务级别的锁,确保同一时间只有一个实例运行
- 有丰富的触发器和作业存储选项:可以使用 ORM、数据库、 Redis 等
- 可持久化任务:当程序重启时,可以持久化存储的任务会再次执行
- 高度可配置:几乎每个组件都可以替换
一个简单的例子:
from apscheduler.schedulers.blocking import BlockingScheduler
def job_function():
print('Hello World!')
scheduler = BlockingScheduler()
scheduler.add_job(job_function, 'interval', seconds=5)
scheduler.start()这个任务会每隔5秒执行一次job_function函数。 更复杂的例子,使用Cron触发器
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
def job_function(): print('Hello World!')
cron_trigger = CronTrigger.from_crontab('*/5 * * * *')
scheduler = BlockingScheduler() scheduler.add_job(job_function, cron_trigger)
scheduler.start()这个任务会每5分钟执行一次,由Cron表达式控制。 apscheduler是一个功能非常强大的任务调度库,可以实现绝大部分定时任务和 cron 任务的需求。
如果你想要给apscheduler定时任务执行的方法传递参数，你可以使用**args或kwargs**参数来指定，例如：
from apscheduler.schedulers.blocking import BlockingScheduler
def my_job(name):
print(f"Hello, {name}")
scheduler = BlockingScheduler()
scheduler.add_job(my_job, "interval", seconds=5, args=["Alice"])
scheduler.add_job(my_job, "interval", seconds=10, kwargs={"name": "Bob"})
scheduler.start()这样，每隔5秒，会执行**my_job("Alice")，每隔10秒，会执行my_job(name="Bob")**。
在django中使用定时任务，并存在数据库中，django重启后继续运行。
- 安装django-apscheduler：pip install django-apscheduler
- 配置django-apscheduler：在settings.py中的INSTALLED_APPS中加入**'django_apscheduler'**，并且确保你已经配置好数据库信息。
- 执行迁移：python manage.py migrate，这会在数据库中生成两个表，分别是django_apscheduler_djangojob和django_apscheduler_djangojobexecution，用来存储任务和执行结果
- 还提供了其他的操作删除任务：scheduler.remove_job(job_name)，暂停任务：scheduler.pause_job(job_name)，开启任务：scheduler.resume_job(job_name)，获取所有任务：scheduler.get_jobs()，修改任务：scheduler.modify_job(job_name) 注：修改任务只能修改参数，如果要修改执行时间的话，有3种方法 第一就把任务删了重新创建， 第二直接操作数据库， 第三用到下面重设任务。
- 重设任务
- scheduler.reschedule_job(job_name)
- scheduler.reschedule_job(job_id=“job1”, trigger=‘interval’, minutes=1)
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events
scheduler = BackgroundScheduler()
#添加一个数据库任务存储
scheduler.add_jobstore(DjangoJobStore(),'default')
# 注册事件监听器
register_events(scheduler)
# 启动调度器
scheduler.start()
如果在使用uwsgi部署报错
raise RuntimeError('The scheduler seems to be running under uWSGI, but threads have ' RuntimeError: The scheduler seems to be running under uWSGI, but threads have been disabled. You must run uWSGI with the --enable-threads option for the scheduler to work.
这意味着你的应用程序需要使用线程，但是uWSGI默认没有初始化Python的GIL（全局解释器锁），导致你的线程无法运行。这样做是出于性能的考虑，因为初始化GIL会带来一些开销。
为了解决这个问题，你需要在运行uWSGI时添加**–enable-threads选项，来启用线程支持例如：
uwsgi --http :9090 --wsgi-file foobar.py --enable-threads或者，你可以在uWSGI的配置文件中添加enable-threads = true这一行，来启用线程支持。例如：
[uwsgi]
socket = 127.0.0.1:3034
plugins-dir = /srv/uwsgi
plugin = python34
uid = py3utils
gid = py3utils
chdir = /srv/python/3/py3utils/tht/app/
wsgi-file = wsgi.py
enable-threads = true
daemonize = % (chdir)/../uwsgi.log
master = true
die-on-term = true
touch-reload = ../uwsgi_restart.txt