Django==2.0.2 Celery==4.4.2 Redis==3.0.504

配置環境

windows10 Django2.0.2 Celery4.4.2 Redis==3.0.504 eventlet(windows環境下安裝，linux環境下不需要，版本随意)

settings 同級目錄下建立 celery.py

# -*- coding:utf-8 -*-

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings

# 指定Django預設配置檔案子產品
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'xrtProject.settings')

# 為我們的項目myproject建立一個Celery執行個體。這裡不指定broker backend 容易出現錯誤。
# 如果有密碼 使用 'redis://:[email protected]:6379/0'

app = Celery('xrt', broker='redis://localhost:6379/10',
             backend='redis://localhost:6379/9')
# 這裡指定從django的settings.py裡讀取celery配置

# app.config_from_object('django.conf:settings')

from kombu import serialization
serialization.registry._decoders.pop("application/x-python-serialize")

# celery 的配置項目，有其他的要求，可以按需求添加
app.conf.update(
    CELERY_ACCEPT_CONTENT = ['json'],
    CELERY_TASK_SERIALIZER = 'json',
    CELERY_RESULT_SERIALIZER = 'json',
)

# 自動從所有已注冊的django app中加載任務
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

           

應用app 中建立 tasks.py

# -*- coding:utf-8 -*-
from __future__ import absolute_import
import json
from celery import shared_task
from django_redis import get_redis_connection

from src.remote.models import OperationLog
from tools.add_message import SendMessage

conn = get_redis_connection('default')


# # name表示設定任務的名稱，如果不填寫，則預設使用函數名做為任務名
# 定義的兩個異步函數


@shared_task(name="save_message", ignore_result=True)
def save_message(phone, recipient, remark, content, sup_links, links, name, remind):
    """存儲全局消息資訊"""
    print("異步任務開始---------------》")
    send_ = SendMessage()
    send_.send_duan_xin(phone, content)
    send_.send_web(content, recipient, sup_links, links, remark, name, remind)
    openid = ''
    send_.send_we_chat(openid, content)
    print("異步任務完成---------------》")
    return "save_message"


@shared_task(name='save_log', ignore_result=True)
def save_log(create_id, create_user, mod_f, mod_s, order, detail, feed_b, created_time):
    print(create_id, create_user, mod_f, mod_s, order, detail, feed_b, created_time)
    dic = {
        "create_id": create_id,
        "create_user": create_user,
        "mod_f": mod_f,
        "mod_s": mod_s,
        "order": order,
        "detail": detail,
        "feed_b": feed_b,
        "created_time": created_time,
    }
    OperationLog.objects.create(**dic)
    print("異步任務完成---------------》")
    return "save_message"


           

在views.py中進行調用

class Log(MyBaseView):
    """日志管理"""
    model = OperationLog

    def post(self, request):
         phone = "138****0342"
         recipient = "1111111111111111111111"
         remark = ''
         content = "您收到一條*********"
         sup_links = "立即檢視"
         links = 'https://www.baidu.com/{}'.format("111111111")
         name = '申請'
         remind = ['短信', '小鈴铛', '公衆号']
        
         tasks.save_message.apply_async(
             (phone, recipient, remark, content, sup_links, links, name, remind))  # 異步調用發送消息
	     return JsonResponse({'code': 200, 'total': total, 'data': rows,  'currentPage': page_number},
	                  json_dumps_params={'default': mydefault})

           

啟動

啟動Django項目

python manage.py runserver
           

啟動Celery

異步任務啟動
	windows 10 環境下：（pip install eventlet）
 		>celery -A devices_back worker -l info -P eventlet
 	linux 下：
 		>celery - A devices_back  worker - l info