apscheduler 定时任务以及存储数据库使用

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**选项，来启用线程支持1 2。例如：

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

读者评论