Django+celery+flower
- Django的定时任务及可视化监控
- Django
Django的定时任务及可视化监控
Django的定时任务，以及可视化监控。
Django
- Django；
首先在python中新建虚拟环境并激活
pip install virtualenv
python -m venv venv
source venv/bin/activate
pip install django #安装django
安装
- 创建django项目；
django-admin startproject myproject
cd myproject
启动Django开发服务器：
python manage.py runserver
默认情况下，服务器会运行在 http://127.0.0.1:8000/。
完成后，可以停用虚拟环境：
deactivate
- redis；
这里使用源码安装，参考redis官网
新开一个窗口，不在python虚拟环境操作
wget https://download.redis.io/redis-stable.tar.gz
tar -xzvf redis-stable.tar.gz
cd redis-stable
make
sudo yum install openssl-devel
make BUILD_TLS=yes
sudo make install
cp redis.conf /etc/
redis-server /etc/redis.conf
检查redis是否启动
redis-cli
127.0.0.1:6379> ping
PONG #成功
- celery简单示例；
安装
pip install celery django-celery-beat
新建一个应用
python manage.py startapp myapp
将 myapp 和 django_celery_beat 添加到 INSTALLED_APPS 中，编辑 myproject/settings.py 文件：
INSTALLED_APPS = [
...
'django_celery_beat',
'myapp',
]
在 myproject 文件夹下创建一个名为 celery.py 的文件，并添加以下内容：
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
在 myproject/init.py 文件中添加以下内容：
from .celery import app as celery_app
__all__ = ('celery_app',)
在 myproject/settings.py 文件中添加 Celery 配置：
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_TIMEZONE = "Asia/Shanghai"
CELERY_ENABLE_UTC = False
在 myapp 文件夹下创建一个名为 tasks.py 的文件，并定义一个简单的定时任务：
from celery import shared_task
import datetime
@shared_task
def print_time():
now = datetime.datetime.now()
print(f"Current time: {now}")
在 myproject/celery.py 文件中定义定时任务的调度计划：
from celery.schedules import crontab
app.conf.beat_schedule = {
'print-time-every-minute': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute='*'),
},
}
这里定义了一个名为 print-time-every-minute 的定时任务，它会每分钟执行一次 myapp.tasks.print_time 任务。
启动 Celery worker：
celery -A myproject worker -l info
启动 Celery beat：
celery -A myproject beat -l info
启动 Django 开发服务器：
python manage.py runserver
查看结果
在终端中可以看到 Celery beat 每分钟调度一次 print_time 任务，并在 Celery worker 的终端中打印当前时间。
- 定时任务时间写法；
在 Celery 中，定时任务的定时时间可以通过多种方式指定，主要包括以下几种常见的写法
1)使用 crontab,例如每分钟、每小时、每天、每周等
from celery.schedules import crontab
app.conf.beat_schedule = {
'print-time-every-minute': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute='*'), # 每分钟执行一次
},
'print-time-every-hour': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=0, hour='*'), # 每小时的第0分钟执行一次
},
'print-time-every-day': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=0, hour=0), # 每天的凌晨0点执行一次
},
'print-time-every-week': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=0, hour=0, day_of_week=1), # 每周一凌晨0点执行一次
},
}
参数说明：
minute：分钟，范围是 0-59。
hour：小时，范围是 0-23。
day_of_week：星期几，范围是 0-6，其中 0 表示星期天。
2)使用 timedelta,指定任务的执行间隔，例如每几秒、几分钟、几小时等。
from datetime import timedelta
app.conf.beat_schedule = {
'print-time-every-10-seconds': {
'task': 'myapp.tasks.print_time',
'schedule': timedelta(seconds=10), # 每10秒执行一次
},
'print-time-every-15-minutes': {
'task': 'myapp.tasks.print_time',
'schedule': timedelta(minutes=15), # 每15分钟执行一次
},
'print-time-every-2-hours': {
'task': 'myapp.tasks.print_time',
'schedule': timedelta(hours=2), # 每2小时执行一次
},
}
3)timedelta 的固定时间间隔,需要任务在固定的时间间隔内执行，可以使用 timedelta 的组合
from datetime import timedelta
app.conf.beat_schedule = {
'print-time-every-1-hour-30-minutes': {
'task': 'myapp.tasks.print_time',
'schedule': timedelta(hours=1, minutes=30), # 每1小时30分钟执行一次
},
'print-time-every-2-days-12-hours': {
'task': 'myapp.tasks.print_time',
'schedule': timedelta(days=2, hours=12), # 每2天12小时执行一次
},
}
4)crontab 的高级组合,支持更复杂的组合，例如指定特定的分钟、小时、星期几等
from celery.schedules import crontab
app.conf.beat_schedule = {
'print-time-specific-time': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=30, hour=14, day_of_week='mon,tue,wed'), # 每周一、二、三的下午2点30分执行一次
},
'print-time-specific-dates': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=0, hour=0, day_of_month='1,15'), # 每月的1号和15号凌晨0点执行一次
},
}
参数说明：
minute：可以指定具体的分钟，例如 30 或 0,15,30,45。
hour：可以指定具体的小时，例如 14 或 0-23。
day_of_week：可以指定具体的星期几，例如 mon,tue,wed 或 1-5。
day_of_month：可以指定具体的日期，例如 1,15。
5)crontab 的通配符指定更灵活的时间。
from celery.schedules import crontab
app.conf.beat_schedule = {
'print-time-every-other-minute': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute='*/2'), # 每隔2分钟执行一次
},
'print-time-every-other-hour': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=0, hour='*/2'), # 每隔2小时执行一次
},
'print-time-every-weekday': {
'task': 'myapp.tasks.print_time',
'schedule': crontab(minute=0, hour=0, day_of_week='1-5'), # 每个工作日的凌晨0点执行一次
},
}
参数说明：
minute=‘/2’：每隔2分钟执行一次。
hour='/2’：每隔2小时执行一次。
day_of_week=‘1-5’：表示星期一到星期五。
6)timedelta 的动态间隔,动态调整任务的执行间隔，可以在任务中动态修改 beat_schedule
from datetime import timedelta
from celery import shared_task
@shared_task
def dynamic_task():
print("Dynamic task executed")
# 修改任务的执行间隔
app.conf.beat_schedule['dynamic-task']['schedule'] = timedelta(seconds=30)
app.conf.beat_schedule = {
'dynamic-task': {
'task': 'myapp.tasks.dynamic_task',
'schedule': timedelta(seconds=10), # 初始间隔为10秒
},
}
- flower；
安装flower
pip install flower
启动，启动之前确保celery已经配置好并运行正常
celery -A myproject flower --port=5555 --broker=redis://localhost:6379/0
加密启动
celery -A myproject flower --port=5555 --broker=redis://localhost:6379/0 --basic_auth=user:password
配置celery，在 Celery 配置中启用事件发送功能，在 myproject/celery.py 文件中添加以下配置：
app.conf.update(
worker_send_task_events=True, # 启用任务事件发送
task_send_sent_event=True, # 启用任务发送事件
)
在浏览器中访问 http://127.0.0.1:5555/，即可看到 Flower 的监控界面
