首页 - 互联网

Celery在Django项目中集成

作者: 五速梦信息网
时间: 2026年04月04日 13:48

使用celery第一件要做的最为重要的事情是需要先创建一个Celery实例对象，我们一般叫做celery应用对象，或者更简单直接叫做一个app。app应用对象是我们使用celery所有功能的入口，比如启动celery、创建任务，管理任务，执行任务等.

celery框架有2种使用方式，一种是单独一个项目目录，另一种就是Celery集成到web项目框架中。

Celery作为单个项目运行

项目目录

-- celeryapp

  -- sms  # 异步任务目录,这里拿发送短信来举例,一个类型的任务就一个目录

    - __init__.py

    - ronglianyunapi.py  # 短信发送文件

    - tasks.py  # 任务的文件，文件名必须是tasks.py!!!每一个任务就是一个被装饰的函数，写在任务文件中

  - __init__.py

  - main.py  # 入口程序

  - settings.py  # 配置文件

main.py，代码：

from celery import Celery
# 实例化celery应用，参数一般为项目应用名

app = Celery("lg_shop")
# 通过app实例对象加载配置文件(参数为导包路径)

app.config_from_object("celeryapp.settings")
# 注册任务， 自动搜索并加载任务

# 参数必须必须是一个列表，里面的每一个任务都是任务的路径名称

# app.autodiscover_tasks(["任务1","任务2",....])

app.autodiscover_tasks(["celeryapp.sms"])
# 启动Celery的终端命令

# celery -A celeryapp.main worker --loglevel=info  # windows不好使，能接收到任务，但不执行

# celery -A celeryapp.main worker --loglevel=info --pool=solo  # windows加上这个参数

# celery -A celeryapp.main worker --loglevel=info -P eventlet -c 10  # 或者使用eventlet，-c是协程的数量，生产环境可以用1000

配置文件settings.py，代码：

REDIS_PASSWORD = "root123456"

REDIS_HOST = "127.0.0.1"

REDIS_PORT = 6379
# 任务队列的链接地址

broker_url = 'redis://:%s@%s:%s/14' % (

    REDIS_PASSWORD, REDIS_HOST, REDIS_PORT

)

# 结果队列的链接地址

result_backend = 'redis://:%s@%s:%s/15' % (

    REDIS_PASSWORD, REDIS_HOST, REDIS_PORT

)

sms/tasks.pytasks.py

from ..main import app

from . import sms_sington
@app.task(name="send_sms_code")

def send_sms_code(tid, mobile, datas):

    response = sms_sington.send_sms(tid=tid, mobile=mobile, datas=datas)

    return response
@app.task(name="send_sms1")

def send_sms1():

    """没有任何参数的异步任务"""

    print('任务：send_sms1执行了...')
@app.task(name="send_sms2")

def send_sms2(mobile, code):

    """有参数的异步任务"""

    print(f'任务：send_sms2执行了...mobile={mobile}, code={code}')
@app.task(name="send_sms3")

def send_sms3():

    """有结果的异步任务"""

    print('任务：send_sms3执行了...')

    return 100
@app.task(name="send_sms4")

def send_sms4(x, y):

    """有结果的异步任务"""

    print('任务：send_sms4执行了...')

    return x + y

sms/ronglianyunapi.py

import os

import json
from ronglian_sms_sdk import SmsSDK
RONGLIANYUN = {

    "accId": '2c94811c86c00e9b0186f2873a040afa',

    "accToken": os.environ.get("RONGLIANYUNACCTOKEN"),

    "appId": '2c94811c86c00e9b0186f2873b0d0b01',

    "reg_tid": 1,  # 注册短信验证码的模板ID

    "sms_expire": 60,  # 短信有效期，单位：秒(s)

    "sms_interval": 60,  # 短信发送的冷却时间，单位：秒(s)

}
class SMSSington:

    def __new__(cls, *args, **kwargs):

        if not hasattr(cls, "_instance"):

            cls._instance = super().__new__(cls, *args, **kwargs)

            ronglianyun = RONGLIANYUN

            cls._instance.sms = SmsSDK(ronglianyun.get("accId"), ronglianyun.get("accToken"), ronglianyun.get("appId"))

        return cls._instance
def send_sms(self, mobile, datas, tid=1):

        """

        发送短信

        @params tid: 模板ID，默认测试使用1

        @params mobile: 接收短信的手机号，多个手机号使用逗号隔开

                单个号码： mobile="13312345678"

                多个号码： mobile="13312345678,13312345679,...."

        @params datas: 短信模板的参数列表(测试只能发四位验证码)

                例如短信模板为： 【云通讯】您的验证码是{1}，请于{2}分钟内正确输入。

                则datas=("1234",5,)

        """

        sdk = self._instance.sms

        resp = sdk.sendMessage(tid, mobile, datas)

        response = json.loads(resp)

        return response

sms/__init__.pytasks.py

from .ronglianyunapi import SMSSington
sms_sington = SMSSington()

接下来，我们运行celery。

# 项目根目录下运行
# 启动Celery的终端命令（普通的运行方式）

# celery -A celeryapp.main worker --loglevel=info  # windows不好使，能接收到任务，但不执行

# celery -A celeryapp.main worker --loglevel=info --pool=solo  # windows加上这个参数

# celery -A celeryapp.main worker --loglevel=info -P eventlet -c 10  # 或者使用eventlet，-c是协程的数量，生产环境可以用1000
# 启动多工作进程，以守护进程的模式运行[一个工作进程就是4个子进程]

# windows大概率报错

# 注意：pidfile和logfile必须以绝对路径来声明

celery multi start worker -A celeryapp.main -l info -n worker1

celery multi start worker -A celeryapp.main -l info -n worker2
# 关闭运行的工作进程

celery multi stop worker -A celeryapp.main -l info -n worker1

celery multi stop worker -A celeryapp.main -l info -n worker2

调用上面的异步任务，拿django的shell进行举例：

# 终端调试

python manage.py shell
# 调用celery执行异步任务

from celeryapp.sms.tasks import send_sms1,send_sms2,send_sms3,send_sms4

mobile = "18533538210"

code = "6666"
# delay 表示马上按顺序来执行异步任务，在celrey的worker工作进程有空闲的就立刻执行

# 可以通过delay异步调用任务，可以没有参数

ret1 = send_sms1.delay()

# 可以通过delay传递异步任务的参数，可以按位置传递参数，也可以使用命名参数

# ret2 = send_sms.delay(mobile=mobile,code=code)

ret2 = send_sms2.delay(mobile,code)
# apply_async 让任务在后面指定时间后执行，时间单位：秒/s

# 任务名.apply_async(args=(参数1,参数2), countdown=定时时间)

ret4 = send_sms4.apply_async(kwargs={"x":10,"y":20},countdown=30)
# 根据返回结果，不管delay，还是apply_async的返回结果都一样的。

ret4.id      # 返回一个UUID格式的任务唯一标志符，5b2becb8-c1ba-4e33-8593-9c6fa7ff8c78

ret4.status  # 查看当前任务的状态 SUCCESS表示成功！ PENDING任务等待

ret4.get()   # 获取任务执行的结果[如果任务函数中没有return，则没有结果，如果结果没有出现则会导致阻塞]
if ret4.status == "SUCCESS":

    print(ret4.get())

在django的视图里面，我们调用Celery来异步执行任务（这里容联云发送短信举例）。

***/views.py

# 异步发送短信

from celeryapp.sms.tasks import send_sms_code

# 原来的

response = send_sms_code(tid=settings.RONGLIANYUN.get("reg_tid"), mobile=mobile,

                                       datas=(sms_code, SMS_CODE_EXPIRES // 60))

# 改成

response = send_sms_code.delay(tid=settings.RONGLIANYUN.get("reg_tid"), mobile=mobile,

                                       datas=(sms_code, SMS_CODE_EXPIRES // 60))

上面就是使用celery并执行异步任务的第一种方式，适合在一些无法直接集成celery到项目中的场景，或是想celery单独跑在一台服务器上的场景。

Celery作为第三方模块在Django中集成

这里还是拿django来举例，目录结构调整如下：

-- lg_shop/

  -- lg_shop/

    -- apps/

      -- verify/

        - tasks.py  # 分散在各个子应用下的异步任务模块

    -- libs/

      -- sms/

        - __init__.py

        - ronglianyunapi.py

    - __init__.py

  - __init__.py

  - celery_app.py  # celery入口程序，相当于上一种用法的main.py

lg_shop/celery_app.py

import os

from celery import Celery
# 必须在实例化celery应用对象之前执行(Django项目manage.py里面的)

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'lg_shop.settings.dev')
# 实例化celery应用，参数一般为项目应用名

app = Celery("lg_shop")

# 指定任务的队列名称

app.conf.task_default_queue = 'Celery'

# 可以把配置写在django的项目配置中

# 设置django中配置信息以 "CELERY_"开头为celery的配置信息

app.config_from_object('django.conf:settings', namespace='CELERY')

# 注册任务，自动搜索并加载任务

app.autodiscover_tasks()
# 启动Celery的终端命令

# celery -A lg_shop worker --loglevel=info  # windows不好使，能接收到任务，但不执行

# celery -A lg_shop worker --loglevel=info --pool=solo  # windows加上这个参数

# celery -A lg_shop worker --loglevel=info -P eventlet -c 10  # 或者使用eventlet，-c是协程的数量，生产环境可以用1000

settings/dev.py，django配置中新增celery相关配置信息，代码：

# celery相关

# Celery异步任务队列框架的配置项[注意：django的配置项必须大写，所以这里的所有配置项必须全部大写]

# 任务队列的链接地址

CELERY_BROKER_URL = 'redis://:%s@%s:%s/14' % (

    REDIS_PASSWORD, REDIS_HOST, REDIS_PORT

)

# 结果队列的链接地址

CELERY_RESULT_BACKEND = 'redis://:%s@%s:%s/15' % (

    REDIS_PASSWORD, REDIS_HOST, REDIS_PORT

)

# 时区，与django的时区同步

CELERY_TIMEZONE = TIME_ZONE

# 防止死锁

CELERY_FORCE_EXECV = True

# 设置并发的worker数量

CELERYD_CONCURRENCY = 200

# celery的任务结果内容格式

CELERY_ACCEPT_CONTENT = ['json', 'pickle']

# 设置失败允许重试[这个慎用，如果失败任务无法再次执行成功，会产生指数级别的失败记录]

# CELERY_ACKS_LATE = True

# 每个worker工作进程最多执行500个任务被销毁，可以防止内存泄漏，500是举例，根据自己的服务器的性能可以调整数值

# CELERYD_MAX_TASKS_PER_CHILD = 500

# 单个任务的最大运行时间，超时会被杀死[慎用，有大文件操作、长时间上传、下载任务时，需要关闭这个选项，或者设置更长时间]

# CELERYD_TIME_LIMIT = 10 * 60

# 任务发出后，经过一段时间还未收到acknowledge, 就将任务重新交给其他worker执行

CELERY_DISABLE_RATE_LIMITS = True
# 之前定时任务（定时一次调用），使用了apply_async({}, countdown=30);

# 设置定时任务（定时多次调用）的调用列表，需要单独运行SCHEDULE命令才能让celery执行定时任务：

# celery -A celeryapp.main beat，当然worker还是要启动的
# from celery.schedules import crontab

# CELERY_BEAT_SCHEDULE = {

#     "send_sms1": {  # 定时任务的注册标记符[必须唯一的]

#         "task": "send_sms1",  # 定时任务的任务名称

#         "schedule": 10,  # 定时任务的调用时间，10表示每隔10秒调用一次add任务

#         # "schedule": crontab(hour=7, minute=30, day_of_week=1),  # 定时任务的调用时间，每周一早上7点30分调用一次add任务

#     }

# }

lg_shop/__init__.py

import pymysql
pymysql.install_as_MySQLdb()

from .celery_app import app as celery_app
__all__ = ['celery_app']

***/tasks.pyapp.autodiscover_tasks

from celery import shared_task
from sms import sms_sington
@shared_task(name="send_sms_code")

def send_sms_code(tid, mobile, datas):

    response = sms_sington.send_sms(tid=tid, mobile=mobile, datas=datas)

    return response

libs/sms/__init__.py

from .ronglianyunapi import SMSSington
sms_sington = SMSSington()

libs/sms/ronglianyunapi.py

import json
from django.conf import settings
from ronglian_sms_sdk import SmsSDK
from logger import log

from response_code import RETCODE
class SMSSington:

    def __new__(cls, *args, **kwargs):

        if not hasattr(cls, "_instance"):

            cls._instance = super().__new__(cls, *args, **kwargs)

            ronglianyun = settings.RONGLIANYUN

            cls._instance.sms = SmsSDK(ronglianyun.get("accId"), ronglianyun.get("accToken"), ronglianyun.get("appId"))

        return cls._instance
def send_sms(self, mobile, datas, tid=1):

        """

        发送短信

        @params tid: 模板ID，默认测试使用1

        @params mobile: 接收短信的手机号，多个手机号使用逗号隔开

                单个号码： mobile="13312345678"

                多个号码： mobile="13312345678,13312345679,...."

        @params datas: 短信模板的参数列表(测试只能发四位验证码)

                例如短信模板为： 【云通讯】您的验证码是{1}，请于{2}分钟内正确输入。

                则datas=("1234",5,)

        """

        sdk = self._instance.sms

        resp = sdk.sendMessage(tid, mobile, datas)

        response = json.loads(resp)

        if response.get("statusCode") != "000000":

            msg = {"code": RETCODE.SMSCODESENDRR, "msg": f"{mobile}验证码发送失败-{response.get('statusMsg')}"}

            log.error(msg)

        return response

django中的用户发送短信，就可以改成异步发送短信了。

视图中调用异步发送短信的任务，代码：

# 异步发送短信

from 子应用.tasks import send_sms_code

# 原来的

response = send_sms_code(tid=settings.RONGLIANYUN.get("reg_tid"), mobile=mobile,

                                       datas=(sms_code, SMS_CODE_EXPIRES // 60))

# 改成

response = send_sms_code.delay(tid=settings.RONGLIANYUN.get("reg_tid"), mobile=mobile,

                                       datas=(sms_code, SMS_CODE_EXPIRES // 60))

终端下先启动celery，在django项目根目录下启动。

# 1. 普通运行模式，关闭终端以后，celery就会停止运行

# celery -A 项目名 worker -l info

celery -A lg_shop worker -l info --pool=solo

# 2. 启动多worker进程模式，以守护进程的方式运行，不需要在意终端。但是这种运行模型，一旦停止，需要手动启动。

# windows大概率报错

celery multi start worker -A lg_shop -l info -n lg_shop_worker

# 3. 停止多worker进程模式

celery multi stop worker -A lg_shop -l info -n lg_shop_worker

关于celery中异步任务发布的2个方法的参数如下：

异步任务名.delay(*arg, **kwargs)

异步任务名.apply_async((arg,), {'kwarg': value}, countdown=60, expires=120)

定时任务的调用器启动，可以在运行了worker以后，使用以下命令：

celery -A lg_shop beat

beat调度器关闭了，则定时任务无法执行，如果worker工作进程关闭了，则celery关闭，保存在消息队列中的任务就会囤积在那里。

代码示例

上一篇： celery中异步延迟执行任务apply
下一篇： CDOJ 1280 772002画马尾每周一题 div1 矩阵快速幂中二版

Celery在Django项目中集成

Celery作为单个项目运行

Celery作为第三方模块在Django中集成

代码示例

相关文章

celery中异步延迟执行任务apply

cell 转成矩阵bimatlab

centeros 给用户设置写入权限

CDOJ 1280 772002画马尾每周一题 div1 矩阵快速幂中二版

CCS中CMD文件解析

CCS中cmd文件的编写

最小公倍数 ((LCM))

最新 Sublime Text3 激活码 (Build 3114 有效)

最新Velocity使用和Velocity语法

最新的iOS应用上架App Store详细流程解析

.NET Core 3.0之深入源码理解Startup的注册及运行

作为一个新手的Oracle（DBA）学习笔记【转】

测试文章2

测试文章1

Celery在Django项目中集成

Celery作为单个项目运行

Celery作为第三方模块在Django中集成

代码示例

相关文章

celery中异步延迟执行任务apply

cell 转成 矩阵bimatlab

centeros 给用户设置写入权限

CDOJ 1280 772002画马尾 每周一题 div1 矩阵快速幂 中二版

CCS中CMD文件解析

CCS中cmd文件的编写

cell 转成矩阵bimatlab

CDOJ 1280 772002画马尾每周一题 div1 矩阵快速幂中二版