Django Celery 的配置及使用---最详细教程

Django Celery 的配置及使用

Redis提供队列消息功能

一、安装redis

系统版本：Ubuntu 20.04

1、获取最新软件包

sudo apt update 
sudo apt install redis-server

2、安装完成后，Redis服务器会自动启动。查看redis是否启动成功

sudo systemctl status redis-server 
或者 service redis-server status

二、开启远程服务

1、找到redis配置文件

vim /etc/redis/redis.conf

# 允许任何主机访问
注释掉这行 # bind 0.0.0.0 ::1

允许任何主机访问
protected-mode yes 改为 protected-mode no

2、重启 Redis 服务，使应用生效

sudo systemctl restart redis-server

三、开启默认端口

1、查看已开启的端口

sudo ufw status

2、开启6379端口

sudo ufw allow 6379

或者直接关闭防火墙

ufw  disable   //关闭防火墙
ufw  enable   //开启

3、验证设置是否完成

# 远程机器ping redis服务器
redis-cli -h 10.0.5.17 ping

# 输出PONG就代表欧克
PONG

四、常用命令

启动Redis服务：
sudo systemctl start redis-server 或者 service redis-server start
关闭Redis服务：
sudo systemctl stop redis-server 或者service redis-server stop
重启Redis服务：
sudo systemctl restart redis-server或者service redis-server restart

django-celery的配置

1、安装celery

pip install celery 或者 pip install celery==4.4.2 

2、添加配置（settings.py）

# celery 相关配置
# 配置celery时区，默认时UTC。
CELERY_TIMEZONE = TIME_ZONE

# 任务队列的链接地址 celery配置redis作为队列。redis有16个数据库，编号0~15。
CELERY_BROKER_URL = 'redis://10.0.5.17:6379/2'

# 设置存储结果的后台  结果队列的链接地址
CELERY_RESULT_BACKEND = 'redis://10.0.5.17:6379/3'

# 可接受的内容格式
CELERY_ACCEPT_CONTENT = ["json"]
# 任务序列化数据格式
CELERY_TASK_SERIALIZER = "json"
# 结果序列化数据格式
CELERY_RESULT_SERIALIZER = "json"

3、在项目同名目录下创建celery.py（必须按照要求来）

from __future__ import absolute_import, unicode_literals

import logging
import os
from celery import Celery

# 指定Django默认配置文件模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'RpkiProject.settings')

# 为项目demos创建一个Celery实例。
app = Celery('RpkiProject')

# 这里指定从django的settings.py里读取celery配置,且setting中的配置文件必须以'CELERY_'开头
app.config_from_object('django.conf:settings', namespace='CELERY')

# 自动从所有已注册的django app中加载任务
app.autodiscover_tasks()

# 配置 Celery 日志
# logger = logging.getLogger('celery')
# logger.setLevel(logging.DEBUG)  # 设置日志级别为 DEBUG
#
# # 创建一个文件处理程序，并将其添加到日志
# file_handler = logging.FileHandler('celery.log')
# file_handler.setLevel(logging.DEBUG)
# formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# file_handler.setFormatter(formatter)
# logger.addHandler(file_handler)

4、将程序导入项目同名目录下的init.py中

from __future__ import absolute_import, unicode_literals

from .celery import app as celery_app

__all__ = ('celery_app',)

5、在已经注册app目录中，创建名为tasks.py的文件（如果取别的名字，celery就扫描不到了）

import logging
import sys

from celery import shared_task
from django.http import JsonResponse

@shared_task
def your_async_task(arg1):
    try:
        # 异步任务逻辑
        result = arg1
        return f"The result is: {result}"
    except Exception as e:
        # 处理异常
        raise e

调用异步任务

def some_view(request):
    # 定义参数
    arg1 = [1,2,3]
    arg2 = 10

    # 调用异步任务并传递参数
    result = your_async_task.delay(arg1)
    res = result.ready()
    # res1 = result.traceback()
    return JsonResponse(data=111, safe=False)

6、启动celery

pip install gevent
celery -A RpkiProject worker -l info -P gevent

#后台启动：nohup celery -A RpkiProject worker -l info -P gevent >/dev/null 2>&1 & exit

  nohub celery -A projetname worker -P gevent -c 1000 > celery.log 2>&1 &

  #1. nohub 忽略所有挂断信号(sighup)
  #2. projectname 项目配置目录 
  #3. -P 使用python协程进行任务的开启，开启1000个协程
  #4. celery.log 在当前目录下生成日志文件，也可以使用绝对路径。
  #5. 标准输入输入是文件描述符0，它是命令的输入，缺省是键盘，也可以是文件或其他命令的输出。
  #6. 标准输出输出是文件描述符1，它是命令的输出，缺省是屏幕，也可以是文件。
  #7. 标准错误输入是文件描述符2，这是命令错误的输出，缺省是屏幕，也可以是文件。
  #8. &符号:代表将命令在后台启动。
  #9. 2>&1 将错误输出重定向给标准输出，中间加&f符号。