Django中使用celery分布式处理邮件/信息发送任务

Posted 2021-08-10 大聪明Smart

https://liboer.top/articles/detail/celery-use-django/

celery分布式处理

在django博客中配置邮件服务器后，当有用户给我们评论时，会发送邮件提醒。如果在发送邮件时，邮件那边卡住了，或者卡了几秒则会导致django卡住，要么django无法服务，要么就是时间特别长。

生产者消费者模型

django把发邮件这件事，发到一个redis的list中，让他去发。

celery，可以实现分布式的处理任务事件。专注于实时处理的任务队列，同时也支持任务调度。

安装

pip3 install celery

broker-消息传输的中间件，生产者一旦有消息发送，将发至broker;

backend-用于存储消息/任务结果

worker-工作者，消费/执行broker中消息/任务的进程

要不要用celery：是否发生阻塞，它办不到实时响应。

使用

# 使用worker
# task.py文件
from celery import Celery
# app = Celery('libo', broker='redis://:password@127.0.0.1:6379/1')
app = Celery('libo', broker='redis://:@127.0.0.1:6379/1')  # 无密码
# 创建任务函数


@app.task
def task_test(a, b):
    print("task is running...")
    return a+b

# 启动worker
# 在task.py文件同级目录下执行
# celery -A tasks worker --loglevel=info
# 此模式默认为前台启动，终端中会输出相关的日志

# 模拟生产者
# from task import task_test
# task_test.delay()

存储

celery的存储需要借助redis或者mysql

from celery import Celery
# app = Celery('libo', broker='redis://:password@127.0.0.1:6379/1')
app = Celery('libo', 
             broker='redis://:@127.0.0.1:6379/1',
            backend='redis://:@127.0.0.1:6379/2')  # 无密码
@app.task
def task_test(a, b):
    print("task is running...")
    return a+b
# 模拟生产者
# from task import task_test
# s = task_test.delay(10, 100)
# s.result

Django中使用celery

1. 创建celery配置文件

   项目同名目录下创建celery.py

2. 应用下创建tasks.py集中定义对应worker函数

3. 视图函数充当生产者，推送具体worker函数

4. 项目目录下启动worker

   celery -A 项目同名目录名 worker -l info

# celery.py和setting.py同级目录下
from celery import Celery
from django.conf import settings
import os

# linux下要加一个环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'BlogLee.settings')

app = Celery('bloglee')
app.conf.update(
    BROKER_URL='redis://:@127.0.0.1:6379/1',
)

# 自动去注册的app中发现加载worker函数
app.autodiscover_tasks(settings.INSTALLED_APPS)


# tasks.py在你的应用根目录下
from django.core.mail import send_mail
from BlogLee.celery import app


@app.task
def my_send_mail(title, email_content, email_from, email_to):
    send_mail(title, email_content, email_from, email_to)

    
    
# views.py中需要发送邮件的地方
my_send_mail.delay(
      title,
      email_content,
      settings.EMAIL_HOST_USER,
      email_to,  # 文章作者邮箱
)



# 正式环境下静默模式启动worker
nohup celery -A /mydata/BlogLee/BlogLee worker -P gevent -c 1000 > celery.log 2>&1 &
# nohup:忽略所有挂断信号，无敌，基本上都不能结束它
# -P gevent -c 1000  携程  开1000个
# > celery.log 日志输出到哪里
# 2>&1  在日志文件中正常输出报错信息 print 等
# &：将命令在后台执行