知识星球：知识星球 | 深度连接铁杆粉丝，运营高品质社群，知识变现的工具知识星球是创作者连接铁杆粉丝，实现知识变现的工具。任何从事创作或艺术的人，例如艺术家、工匠、教师、学术研究、科普等，只要能获得一千位铁杆粉丝，就足够生计无忧，自由创作。社群管理、内容沉淀、链接粉丝等就在知识星球。
https://articles.zsxq.com/id_5x1m9wdv3e20.html
目录
一.简单使用celery
二.在django中的使用
1.使用异步任务
2 实现定时任务
3 通过Admin 后台管理添加定时任务
三.实现代码实时监控与预警
4 通过Admin查看任务运行情况
5 通过Flower监控celery运行情况
6 任务执行成功或失败告警(邮件)
7 定时爬取技术类文章-邮件通知
安装
一.简单使用celery
1celery概念
Celery是一个基于Python开发的分布式异步消息任务队列，它能够轻松实现任务的异步处理。Celery的主要应用场景包括异步任务（如发送短信/邮件、消息推送、音频处理等）、定时任务（如定时执行爬虫爬取指定内容）以及实现简单的分布式爬虫系统等。Celery在执行任务时需要通过一个消息中间件（Broker）来接收和发送任务消息，以及存储任务结果，常用的消息中间件有RabbitMQ和Redis。
Celery具有以下优点：
-
简单：Celery易于使用和维护，不需要配置文件，配置和使用简单。
-
高可用：当任务执行失败或执行过程中发生连接中断，Celery会自动尝试重新执行任务。
-
快速：单个Celery进程每分钟可处理数以百万计的任务，保持往返延迟在亚毫秒级。
-
灵活：Celery的几乎所有部分都可以扩展或单独使用，各个部分可以自定义。
Celery的架构主要由三部分组成：消息中间件（Broker）、任务执行单元（Worker）和任务执行结果存储（Backend）。Broker用于存放任务，接收任务生产者发来的消息，并将任务存入队列。Worker是执行任务的处理单元，实时监控消息队列，获取队列中调度的任务，并执行它。Backend用于存储任务的执行结果，以供查询。
2 Celery使用场景
# 1 异步任务（函数）-一些耗时操作交给celery执行-视频转码，邮件发送，消息推送# 2 定时任务-定时推送消息，定时爬取一些数据，定时统计一些数据等# 3 延迟任务-提交任务后，等一段时间再执行任务
3 Celery架构
# Celery 架构，它采用典型的生产者-消费者模式，主要由以下部分组成：
# Producer:它负责把任务(发送短信任务，爬虫任务)提交得到broker中
# celery Beat：会读取文件--》周期性的向broker中提交任务
# broker：消息中间件，放任务的地方，celery本身不提供，借助于redis --》rabbitmq
# worker：工人，消费者，负责从消息中间件中取出任务--》执行
# backend ：worker执行完，会有结果，结果存储在backend，celery不提供--》借助于 redis
快速入手
1-编写celery_demo.py
import time
from celery import Celery
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
app = Celery('demo', broker=broker, backend=backend)
@app.task # 被装饰器装饰了，才是celery的任务
def add(a, b):print('a+b的结果是', a + b)time.sleep(1)# 模拟任务耗时return a + b2-提交任务--add_task.py
from celery_demo import add
res=add.delay(7,8) # 只是提交，没有真正执行
print(res)3-redis中可以看到被提交的任务-没执行
、3.-启动worker执行任务
# win: celery -A celery_demo worker -l info -P eventlet
celery -A 有app的py文件的名字 worker -l info -P eventlet
# mac linux: celery -A celery_demo worker -l info
4 Celery包结构
# 后期随着项目越来越大---》task任务越来越多，都写在一个py文件中，就不好了，希望把任务拆分到多个py文件中# 需要使用包结构来管理
项目名├── celery_task # celery包│ ├── __init__.py # 包文件│ ├── celery.py # celery连接和配置相关文件，且名字必须叫celery.py│ └── crawl_task.py # 所有任务函数└── order_task.py # 所有任务函数└── user_task.py # 所有任务函数├── add_task.py # 添加任务└── get_result.py # 获取结果
4.1 创建包：celery_task
celery.py
from celery import Celerybroker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# 任务分到不同的py文件中了
app = Celery('demo', broker=broker, backend=backend,include=['celery_task.crawl_task', 'celery_task.order_task', 'celery_task.user_task'])
**crawl_task.py **
import timefrom .celery import app@app.task
def crawl_baidu():print('====开始爬百度====')time.sleep(2)print('====爬完百度====')return '成功爬取！！'@app.task
def crawl_cnblogs():print('====开始爬cnblogs====')time.sleep(2)print('====爬完cnblogs====')return '成功爬取cnblogs！！'order_task.py
import time
from .celery import app@app.task
def pay_order():print('====开始下单====')time.sleep(5)print('====下单完成====')return '成功下单！！'
**user_task.py **
import time
# . 表示当前路径
from .celery import app@app.task
def send_email(to='306334678@qq.com'):print('====邮件开始发送====')time.sleep(3)print('====邮件发送完成====')return '向%s发送邮件成功！！' % to
4.2 启动别的程序--提交任务--add_task.py
from celery_task.crawl_task import crawl_baidures = crawl_baidu.delay() # crawl_baidu没有参数，这里就不传
print(res) # uuid--》后续通过uuid查询结果是否执行完成 7a789046-4954-4918-b93e-2defc76adc024.3 提交的任务，没执行，放在redis中，可以去redis中看
4.4 启动worker执行任务
# 在包这一层执行，以包名运行
celery -A celery_task worker -l info -P eventlet# worker就会执行咱们提交的任务，执行完，会把结果放在redis中
# 手动去redis中看4.5 使用代码查看结果get_result.py
# 1 拿到app
from celery_task.celery import app
from celery.result import AsyncResult
# 2 知道我们要查的任务 uuid
id = '7d396575-bfb7-45a5-8a0b-3a4767cde886'
if __name__ == '__main__':result = AsyncResult(id=id, app=app)if result.successful():result = result.get()print(result)elif result.failed():print('任务失败')elif result.status == 'PENDING':print('任务等待中被执行')elif result.status == 'RETRY':print('任务异常后正在重试')elif result.status == 'STARTED':print('任务已经开始被执行')
5 异步任务--延迟任务--定时任务
5.1 异步任务
# 咱们刚刚讲的就是在做异步任务res = 任务.delay(10, 20)
5.2 延迟任务
# 延迟5s钟给 616564099@qq.com 发邮件#######提交延迟任务 延迟5s钟给 616564099@qq.com 发邮件
# 创建一个5s时间对象
from celery_task.user_task import send_email
from datetime import datetime,timedelta
# datetime.utcnow() 默认时区用的utc时间，咱们也需要使用utc时间
# 后期我们会换成上海时间
# utc时间5s后的时间
eta=datetime.utcnow() + timedelta(seconds=5)
res=send_email.apply_async(args=['616564099@qq.com'],eta=eta)
print(res)# 启动worker
celery -A celery_task worker -l info -P eventlet
# worker收到后，不会立即执行，等5s后执行# 问题
如果延迟任务提交了，但是worker没启动，等10s后，worker才启动，worker也会执行任务，但是是立即执行# 后期worker会一直在运行，没有任务就阻塞，一般不会让它停掉，咱们这个问题一般不会出现# 加入只有一个worker，进来俩任务，串行执行，可以在多个机器上启动多个worker5.3 定时任务
from celery import Celerybroker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# 任务分到不同的py文件中了
app = Celery('demo', broker=broker, backend=backend,include=['celery_task.crawl_task', 'celery_task.order_task', 'celery_task.user_task'])#1 celery.py中加入
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False# 任务的定时配置
from datetime import timedelta
app.conf.beat_schedule = {'low-task': {'task': 'celery_task.crawl_task.crawl_baidu','schedule': timedelta(seconds=5),'args': (), # crawl_baidu 没有参数，所以把不传}
}二.在django中的使用
1在项目路径下创建 celery.py
import os
from celery import Celery
# 1 导入django的配置文件---》后续在celery的任务中，就可以直接使用django的orm，缓存。。。
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery_crawl.settings')
# 2 实例化得到celery对象
app = Celery('proj')
# 3 celery的配置，使用django 配置文件中的配置--》刚刚写的配置
app.config_from_object('django.conf:settings')
# 4 这句话会去所有app中，自动查找 tasks.py 文件，作为任务文件
app.autodiscover_tasks()
1.2django 的配置文件 settings.py
############# celery的配置信息######
#1 Broker配置，使用Redis作为消息中间件
BROKER_URL = 'redis://127.0.0.1:6379/1'
#2 BACKEND配置，使用redis
RESULT_BACKEND = 'redis://127.0.0.1:6379/2'
#3 序列化方案--》json
ACCEPT_CONTENT = ['json']
TASK_SERIALIZER = 'json'
# 结果序列化方案
RESULT_SERIALIZER = 'json'# 4 任务结果过期时间，秒
TASK_RESULT_EXPIRES = 60 * 60 * 24#5 时区配置
TIMEZONE = 'Asia/Shanghai'
1.3以后只需要再不同app中，写tasks.py
# 注册任务from celery import shared_task# 以后使用shared_task 替换掉 app.task# 任务函数
@shared_task
def add(a, b):return a + b
1.4 在项目目录下的 init.py中加入
from .celery import app as celery_app__all__ = ('celery_app',)
1.5 启动celery的worker-写一个提交异步任务的视图函数
1.使用异步任务
1.5.1 启动worker
# 1 启动worker
celery -A django_celery_crawl worker -l debug -P eventlet1.5.2 写一个视图函数测试
############### 总路由中 urls.py###############
from django.contrib import admin
from django.urls import path,includeurlpatterns = [path('admin/', admin.site.urls),path('app01/', include('app01.urls')), # http://127.0.0.1:8000/app01/send/ 异步发邮件
]###############分路由中 urls.py###############
from django.contrib import admin
from django.urls import path
from .views import send_mail_view
urlpatterns = [path('send/', send_mail_view),
]
###############视图函数中 views.py###############
from django.shortcuts import render, HttpResponse
from .tasks import send_email
def send_mail_view(request):# 取出用户要发送的邮箱to_user = request.GET.get('user')# 异步发送res = send_email.delay(to_user)print(res.id)return HttpResponse('邮件已发送，id号为%s' % res.id)
1.5.3 在浏览器输入：
http://127.0.0.1:8000/app01/send/?user=616564099@qq.com2 实现定时任务
# celery集成到django后，以后定时任务的编写，要写到django的配置文件中
from datetime import timedelta
CELERYBEAT_SCHEDULE = {'every_5_second': {'task': 'app01.tasks.add','schedule': timedelta(seconds=5),'args': (33, 44),}
}
2.1 django的配置文件配置
2.2 重启worker---启动beat
# 启动worker
celery -A django_celery_crawl worker -l debug -P eventlet
# 启动beat
celery -A django_celery_crawl beat -l debug3 通过Admin 后台管理添加定时任务
# 存在问题：如果要再新加一个定时任务（爬美女图片）我们只能修改代码 ：settings.py-->加入代码重启worker，重启beat--》才能行有些麻烦# 通过点点点，就能自动添加定时任务-Admin 后台管理添加定时任务
3.1 引入--安装djiango-celery-beat
pip install django-celery-beat3.2 在app中注册djiango-celery-beat
INSTALLED_APPS = [...'app01','django_celery_beat' # 注册刚刚第三方模块
]3.3 配置django的时区
LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True
USE_L10N = True
USE_TZ = False3.4 在setting中配置调度器
CELERYBEAT_SCHEDULER = 'django_celery_beat.schedulers.DatabaseScheduler'
# 只要配了这个，原来celery中的定时任务统一不能用了，需要我们手动配置了
3.5 数据迁移
#1 django 使用mysql数据库
DATABASES = {'default': {'ENGINE': 'django.db.backends.mysql','NAME': 'django_celery_crawl', # 去mysql中创建数据库'HOST':'127.0.0.1','PORT':3306,'USER':'root','PASSWORD':'123' # 自己写自己的密码}
}
# 2 安装mysqlclient
pip install mysqlclient#3 去mysql 中创建 django_celery_crawl 库#4 django-celery-beat 这个app会产生表--》以后只要咱么在表里添加记录，就能制定定时任务
python manage.py makemigrations
python manage.py migrate
3.6 使用admin后台管理插入数据
# 咱么如果想向mysql数据表中，插入数据-可以使用navicat插入-django提供了一个后台管理--》登录到后台管理，可以图形化界面录入数据-但是需要个账号登录:命令创建账号python manage.py createsuperuser-我创建的是：账号： admin 密码 ： 123456
3.7 美化admin
# 1 simpleui 美化admin
# 2 开源地址
https://gitee.com/tompeppa/simpleui
# 3 文档地址
https://newpanjing.github.io/simpleui_docs/config.html# 4 安装
pip3 install django-simpleui# 5 app中注册
INSTALLED_APPS = ['simpleui',...
]
3.8 手动添加任务
# 1 启动worker，beat
# 启动worker
celery -A django_celery_crawl worker -l debug -P eventlet
# 启动beat
celery -A django_celery_crawl beat -l debug
# 2 在admin中手动添加
三.实现代码实时监控与预警
4 通过Admin查看任务运行情况
# 在控制台监控任务执行情况或者去redis中查看，还不是很方便，最好是能够通过web界面看到任务的执行情况，如有多少任务在执行，有多少任务执行失败了等。
# 这个Celery也是可以做到了，就是将任务执行结果写到数据库中，通过web界面显示出来。
这里要用到django-celery-results插件。
通过插件可以使用Django的orm作为结果存储，这样的好处在于我们可以直接通过django的数据查看到任务状态，同时为可以制定更多的操作4.1 安装 django-celery-results
pip install django-celery-results4.2 app中注册
INSTALLED_APPS = (
...,
'django_celery_results',
)4.3 修改配置文件
# 让之前backend----》结果存储放在redis中----》改到放到数据中---》因为在数据库中的咱们可以通过admin查看
#RESULT_BACKEND = 'redis://127.0.0.1:6379/1'
# 使用使用django orm 作为结果存储
CELERY_RESULT_BACKEND = 'django-db' #使用django orm 作为结果存储
4.4 迁移数据库
# 执行迁移命令
python manage.py makemigrations
python manage.py migrate
5 通过Flower监控celery运行情况
如果不想通django的管理界面监控任务的执行，还可以通过Flower插件来进行任务的监控。Flower的界面更加丰富，可以监控的信息更全
Flower 是一个用于监控和管理 Celery 集群的开源 Web 应用程序。它提供有关 Celery workers 和tasks状态的实时信息
# Flower可以：
1 实时监控celery的Events-查看任务进度和历史记录-查看任务详细信息（参数、开始时间、运行时间等）
2 远程操作-查看workers 状态和统计数据-关闭并重新启动workers 实例-控制工作池大小和自动缩放设置-查看和修改工作实例消耗的队列-查看当前正在运行的任务-查看计划任务（预计到达时间/倒计时）-查看保留和撤销的任务-应用时间和速率限制-撤销或终止任务
3 Broker 监控-查看所有 Celery 队列的统计信息5.1 修改结果存储为redis
RESULT_BACKEND = 'redis://127.0.0.1:6379/2'5.2 安装flower和启动
#1 安装
pip install flower# 2 启动
celery -A django_celery_crawl flower --port-5555#3 浏览器访问：
http://127.0.0.1:5555/# 4 启动 worker，beat，flower-一般先启动flower
6 任务执行成功或失败告警(邮件)
虽然可以通过界面来监控了，但是我们想要得更多，人不可能天天盯着界面看吧，如果能实现任务执行失败或成功就自动发邮件告警就好了。这个Celery当然也是没有问题的6.1 task.py 中加入
# 注册任务
import timefrom celery import shared_task### 写个类，继承 Task，写方法
from celery import Taskfrom django.core.mail import send_mail
from django.conf import settingsclass SendEmailTask(Task):def on_success(self, retval, task_id, args, kwargs):# 任务执行成功，会执行这个info = f'爬虫任务成功-- 任务id是:{task_id} , 参数是:{args} , 执行成功 !'# 发送邮件--》django中发送邮件send_mail('celery监控成功告警', info, settings.EMAIL_HOST_USER, ['616564099@qq.com'])def on_failure(self, exc, task_id, args, kwargs, einfo):# 任务执行失败，会执行这个info = f'爬虫任务失败-- 任务id是:{task_id} , 参数是:{args} , 执行失败，请去flower中查看原因 !'# 发送邮件--》django中发送邮件send_mail('celery监控爬虫失败告警', info, settings.EMAIL_HOST_USER, ['616564099@qq.com'])def on_retry(self, exc, task_id, args, kwargs, einfo):# 任务执行重试，会执行这个print('重试了！！！')# 以后使用shared_task 替换掉 app.task# 任务函数
@shared_task
def add(a, b):return a + b@shared_task
def send_email(to_user):time.sleep(2)return '发送邮件成功:%s' % to_user@shared_task(base=SendEmailTask, bind=True) # 只要执行它成功或失败，都会给 616564099@qq.com发送邮件
def crawl_cnblogs(self):print('爬去cnblogs网站技术博客了')return True
6.2 setting中 邮箱配置
#### 发送邮件配置
EMAIL_HOST = 'smtp.qq.com' # 如果是 163 改成 smtp.163.com
EMAIL_PORT = 465
EMAIL_HOST_USER = '' # 帐号
EMAIL_HOST_PASSWORD = '' # 密码
DEFAULT_FROM_EMAIL = EMAIL_HOST_USER
#这样收到的邮件，收件人处就会这样显示
#DEFAULT_FROM_EMAIL = 'lqz<'306334678@qq.com>'
EMAIL_USE_SSL = True #使用ssl
#EMAIL_USE_TLS = False # 使用tls6.3 在admin中添加任务执行
# 1 admin中添加任务，只执行一次
# 2 执行完了 flower可以监控到
# 3 也能收到邮件通知7 定时爬取技术类文章-邮件通知
# 1 爬取目标网站：https://www.cnblogs.com/-爬取它首页的推荐文章-文章标题-文章作者-文章url地址-文章摘要-每隔一天爬一次-可能会重复--》去重# 2 安装-pip install requests-pip install beautifulsoup47.1 models.py 创建表
from django.db import modelsclass Article(models.Model):title = models.CharField(max_length=64)url = models.CharField(max_length=64)author = models.CharField(max_length=64)desc = models.TextField()# 迁移
python manage.py makemigrations
python manage.py migrate# 数据库才有这个表7.2 在task中写
# 注册任务
import timefrom celery import shared_task### 写个类，继承 Task，写方法
from celery import Taskfrom django.core.mail import send_mail
from django.conf import settingsclass SendEmailTask(Task):def on_success(self, retval, task_id, args, kwargs):# 任务执行成功，会执行这个info = f'爬虫任务成功-- 任务id是:{task_id} , 参数是:{args} , 执行成功 !'# 发送邮件--》django中发送邮件send_mail('celery监控成功告警', info, settings.EMAIL_HOST_USER, ['616564099@qq.com'])def on_failure(self, exc, task_id, args, kwargs, einfo):# 任务执行失败，会执行这个info = f'爬虫任务失败-- 任务id是:{task_id} , 参数是:{args} , 执行失败，请去flower中查看原因 !'# 发送邮件--》django中发送邮件send_mail('celery监控爬虫失败告警', info, settings.EMAIL_HOST_USER, ['616564099@qq.com'])def on_retry(self, exc, task_id, args, kwargs, einfo):# 任务执行重试，会执行这个print('重试了！！！')###爬取技术博客#####
import requests
from bs4 import BeautifulSoup
from .models import Article
from redis import Redis@shared_task(base=SendEmailTask, bind=True) # 只要执行它成功或失败，都会给 616564099@qq.com发送邮件
def crawl_cnblogs(self):# 拿到redis链接conn = Redis(host='127.0.0.1', port=6379)res = requests.get('https://www.cnblogs.com/')soup = BeautifulSoup(res.text, 'html.parser')# 查找：-文章标题-文章作者-文章url地址-文章摘要# 找到所有文章article_list = soup.find_all(name='article', class_='post-item')# 循环一个个文章for article in article_list:title = article.find(name='a', class_='post-item-title').texturl = article.find(name='a', class_='post-item-title').attrs.get('href')author = article.find(name='a', class_='post-item-author').span.textdesc = article.find(name='p', class_='post-item-summary').text.strip()print(f'''文章名字：{title}文章地址：{url}文章作者：{author}文章摘要：{desc}''')# 去重，不能有重复,如果重复了，就不存到数据库中了---》集合可以去重--》redis集合去重-->根据url去重# 只要redis的urls的这个集合中，有了这个地址，我们就不存了--》去重了res = conn.sadd('urls', url)if res: # 返回1 表示放进集合中了，表示集合中没有，存数据库 返回0，表示集合中有过，不存了# 保存到数据库中--》创建一个Article的数据表Article.objects.create(title=title, url=url, author=author, desc=desc)return True
7.3 以后只需要在后台管理添加定时任务
# 比如一天爬一次-所有服务都启动，不用管了安装
# 1 创建python项目-pycharm创建
# 2 创建虚拟环境--》解释器中，目前没有任何模块--》后续咱们需要装各种模块-pycharm创建# 3 安装celery
pip install celery# 4 安装redis
pip install redis# 5 如果win平台，需要装
pip install eventlet # celery 是个小组织，它不支持win，借助于eventlet，可以运行在win上
#官方文档：https://docs.celeryq.dev/en/stable/django/first-steps-with-django.htmlpip install Django==3.2.22
pip install celery
pip install redis # celery 的中间件借助于redis--》redis模块
pip install eventlet # 在windows环境下需要安装eventlet包# 创建django项目：名字叫 django_celery_crawl
# 保证django能运行
pip install django-celery-beat
pip3 install django-simpleui
pip install django-celery-results
pip install flower