Django+Celery 執行異步任務和定時任務

celery是一個基於python開發的簡單、靈活且可靠的分布式任務隊列框架，支持使用任務隊列的方式在分布式的機器/進程/線程上執行任務調度。采用典型的生產者-消費者模型，主要由三部分組成:

1. 消息隊列broker：broker實際上就是一個MQ隊列服務，可以使用redis、rabbitmq等作為broker
2. 處理任務的消費者workers：broker通知worker隊列中有任務，worker去隊列中取出任務執行，每一個worker就是一個進程
3. 存儲結果的backend：執行結果存儲在backend，默認也會存儲在broker使用的MQ隊列服務中，也可以單獨配置用何種服務做backend

flask,django是同步框架，所有的請求以隊列形式完成。這樣的話效率極差，用戶體驗不好，為了解決這個問題引入celery異步方式在后台執行這些任務（這里使用到了redis,3.0以下兼容性更好）

1，安裝依賴

pip install celery

pip install celery-with-redis

pip install django-celery

2，settings.py設置

#配置celery
import djcelery
djcelery.setup_loader()
BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_IMPORTS = ('mymac.tasks') #需執行異步的子應用

#將djcelery安裝到應用中
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'mysg',
    'lianxi',
    "rest_framework",
    'corsheaders',
    #異步
    'djcelery',
]

3,將異步的應用中注冊celery.py

import os
import django
from celery import Celery
from django.conf import settings 
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mymac.settings')
django.setup()
app = Celery('mymac')
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

4，建立異步任務和定時任務的(tasks.py)

• 注意tasks.py必須建在各app的根目錄下，且只能叫tasks.py，不能隨意命名

  異步任務

import time,random
from celery import task
#發郵件
from django.core.mail import send_mail
from django.http import HttpResponse
#定義異步寫文件方法
@task
def file_task():
        #寫文件操作  文件對象
        file_object = open("./data.text",'a+',encoding='utf-8')
        file_object.write("hello")
        file_object.close()
        print("ok")

#定義異步發郵件的方法
@task
def email_():
        captcha_text = []
        for i in range(4):
        #定義驗證碼字符
                str = 'qwertyuiopasdfghjklzxcvbnm1234567890'
                c = random.choice(str)
                captcha_text.append(c)
        #返回隨機生成的字符串         
        captcha = "".join(captcha_text)
        res = send_mail("歡迎注冊",'您的驗證碼是:'+ captcha,['[e-mail]'],DEFAULT_FROM_EMAIL)
        if res:
                return HttpResponse("發送成功")
        else:
                return HttpResponse("發送失敗")

   views.py中引用使用這個tasks異步處理

from mymac.tasks import email,file_task

#異步發郵件
def email(request):
    print(email_.delay())
    return HttpResponse("異步發郵件")
#異步寫入文件
def failtask(request):
    print(file_task.delay())
    return HttpResponse("success")

#配好路由觸發任務即可

　　定時任務

#導入定時任務庫
from celery.decorators import periodic_task
from celery.schedules import crontab
#發短信
from twilio.rest import Client

#定義20點10分發送
#@periodic_task(run_every=crontab(minute=10,hour=20))
#定義10秒發送一次
@periodic_task(run_every=10)
def mail():   
        #定義短信sid
        account_sid = 'ACbccc4d2127e888e6f6654dc8128c019e'
        #定義秘鑰
        auth_token = 'a0f31b24c76c65c20c6400dc94537ac6'
        #定義客戶端對象  
        clinet = Client(account_sid,auth_token)
        #定義短信內容 1,發給誰 2,發信人 3,內容
        status = clinet.messages.create(to="[phone]",from_="[phone]",body="hello world")
        if status:
                print("發送成功")

#注意時區 例如中國 
#settings.py 設置  語言相關配置
LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'

crontab的參數有：
month_of_year：月份
day_of_month：日期
day_of_week：周
hour：小時
minute：分鍾

5，啟動任務

啟動服務的命令：

celery -A mymac beat -l info 定時任務
celery -A mymac worker -l info 異步任務