Django中使用celery

项目目录结构

# 创建django项目
django-admin startproject django_celery
# pycharm打开django_celery项目，创建app
python manage.py startapp myapps

配置文件

1
2
3

# mycelery/config.py
broker_url = 'redis://:xxx@111@43.254.x.x:6379/14'
result_backend = 'redis://:xxx@111@43.254.x.x:6379/15'

任务文件

# mycelery/sms/tasks.py
from mycelery.main import app
import time

import logging
log = logging.getLogger("django")

@app.task  # name表示设置任务的名称，如果不填写，则默认使用函数名做为任务名
def send_sms(mobile):
    """发送短信"""
    print("向手机号%s发送短信成功!"%mobile)
    time.sleep(5)
    return "send_sms OK"

@app.task  # name表示设置任务的名称，如果不填写，则默认使用函数名做为任务名
def send_sms2(mobile):
    print("向手机号%s发送短信成功2!" % mobile)
    time.sleep(5)
    return "send_sms2 OK"

主程序

# mycelery/main.py
import os
from celery import Celery
# 创建celery实例对象
app = Celery("djcelery")

# 把celery和django进行组合，识别和加载django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celeryPros.settings.dev')

# 通过app对象加载配置
app.config_from_object("mycelery.config")

# 加载任务
# 参数必须必须是一个列表，里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["mycelery.sms",])

# 启动Celery的命令
# 强烈建议切换目录到mycelery同级目录下启动
# celery -A mycelery.main worker --loglevel=info

django视图

# django_celery/urls.py
from django.urls import path
from myapps import views

urlpatterns = [
    path('admin/', admin.site.urls),
    path('test/', views.test),
]

# myapps/views.py
from django.shortcuts import render,HttpResponse
from mycelery.sms.tasks import send_sms,send_sms2
from datetime import datetime,timedelta

# Create your views here.
def test(request):
    # 异步任务
    # 声明一个和celery一样的任务函数，可以用导包来解决
    # send_sms.delay("111")
    # send_sms2.delay("112")

    # 定时任务
    ctime = datetime.now()
    # 默认用utc时间
    utc_time = datetime.utcfromtimestamp(ctime.timestamp())
    time_delay = timedelta(seconds=10)
    task_time = utc_time + time_delay
    result = send_sms.apply_async(["113",], eta=task_time)
    print(result.id)

    return HttpResponse("ok")

总结

1 2	访问127.0.0.1/test 即可执行myapps的视图的test函数，执行celery任务。执行test函数是异步的，不会等celery任务执行完才返回OK