Django+celery+flower

Django的定时任务及可视化监控

Django的定时任务，以及可视化监控。

Django

1. Django；
   首先在python中新建虚拟环境并激活

pip install virtualenv
python -m venv venv
source venv/bin/activate
pip install django #安装django

安装

2. 创建django项目；

django-admin startproject myproject
cd myproject

启动Django开发服务器：

python manage.py runserver

默认情况下，服务器会运行在 http://127.0.0.1:8000/。

完成后，可以停用虚拟环境：

deactivate

3. 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  #成功

4. 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 Celeryos.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 crontabapp.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 的终端中打印当前时间。

6. 定时任务时间写法；
   在 Celery 中，定时任务的定时时间可以通过多种方式指定，主要包括以下几种常见的写法
   1)使用 crontab,例如每分钟、每小时、每天、每周等

from celery.schedules import crontabapp.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 timedeltaapp.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 timedeltaapp.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 crontabapp.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 crontabapp.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秒},
}

7. 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 的监控界面