異步任務的搭建
?flask與Django配置有差異，這里使用Django，

?

創建異步文件夾celery_task,和以下目錄文件

__init__文件配置

from celery import Celery from datetime import timedeltaimport os import django# 調用項目環境 os.environ.setdefault("DJANGO_SETTINGS_MODULE","s1.settings") django.setup()# 異步任務celery_app = Celery("mycelery") # 創建一個異步任務 celery_app.config_from_object('celery_task.config') # 讀取config里的配置# 定時任務celery_app.autodiscover_tasks(['celery_task.times'])# 文件夾名.文件名celery_app.conf.update(CELERYBEAT_SCHEDULE={'sum-task': {# 文件夾名.文件名.方法名'task': 'celery_task.times.my_con',# 多少秒執行一次（eg:5秒一執行）'schedule': timedelta(seconds=5),# 這是傳入my_con方法中的參數'args': (5, 6)},'sum-task1': {'task': 'celery_task.times.my_print','schedule': timedelta(seconds=3),'args': ()},} )

config.py文件配置

# (注釋部分僅為查詢到的資料，未測試使用，僅供參考)# 如果使用 redis 作為中間人——任務隊列（消息隊列）———作用：暫存任務的地方 # 需要這樣配置: # broker_url = 'redis://192.168.203.153:6379/3' # 如果需要密碼的話 # BROKER_URL='redis://:xxxxx@127.0.0.1:6379/2'# 如果使用別的作為中間人, 例如使用 rabbitmq # 則 rabbitmq 配置如下: # broker_url= 'amqp://用戶名:密碼@ip地址:5672'# 例如: # meihao: 在rabbitq中創建的用戶名, 注意: 遠端鏈接時不能使用guest賬戶. # 123456: 在rabbitq中用戶名對應的密碼 # ip部分: 指的是當前rabbitq所在的電腦ip # 5672: 是規定的端口號 # broker_url = 'amqp://meihao:123456@172.16.238.128:5672'broker_url = 'redis://127.0.0.1:6379/0' # 結果隊列的鏈接地址 result_backend = 'redis://127.0.0.1:6379/1'# 時區 # CELERY_TIMEZONE = 'Asia/Shanghai' timezone = 'Asia/Shanghai'# 導入指定的任務模塊 imports = ('celery_task.task' )

task.py文件配置

# 自定義異步方法import time import random import redisfrom celery_task import celery_app from ronglian_sms_sdk import SmsSDK from rest_framework.response import Response from django.core import mail# 定義發送短信方法 accId="8aaf07087de13e4***********" accToken="419de3ed6144***********" appId ="8aaf07087de13************"# @app.task 指定將這個函數的執行交給celery異步執行 @celery_app.task() def send_mess(phone):'''生成短信驗證碼'''sdk = SmsSDK(accId, accToken, appId)tid = '1' # 短信驗證碼模板# 驗證碼隨機生成datas = (random.randint(000000, 999999),)print("生成的短信驗證碼>>>", datas)# 短信驗證碼的模板 手機號 驗證碼隨機生成resp = sdk.sendMessage(tid, phone, datas)# print("驗證碼的內容>>>",resp)# 存入redis庫中rdb = redis.Redis(host='localhost', port=6379, db=0)# setex (key,秒，驗證碼)if rdb.get('sms_%s' % phone):return Response({'msg': "短信已發送，不能重復發送", 'code': 500})rdb.setex("sms_%s" % phone, 300, datas[0])rdb.close()return resp# 定義發送郵箱方法 @celery_app.task(name='send_email1') def send_email1(subject, message, from_email, to_email):time.sleep(10)mail.send_mail(subject, message, from_email, to_email)print('發郵件')

times.py文件配置

# 自定義定時任務，調用在__init__里配置，#簡單定義測試方法from celery_task import celery_app @celery_app.task() def my_con(x,y):print(x,y)print('x和y相加')return x+y@celery_app.task() def my_print(): print('22222222')

調用命令

# 啟動異步任務

# celery -A 任務名 worker -l info -P eventlet
# celery -A celery_task worker -l info -P eventlet

# 啟動定時任務

# celery -A 任務名 beat -l info
# celery -A celery_task beat -l info

?注意事項

# 注意：任務需要在終端啟動后才可以使用
# 注意：要先啟動異步任務才能啟動定時任務

# 注意：在導入方法,并調用時，需要在方法名后加.delay,加上才是異步調用執行
# 例：調用發送手機短信驗證碼

# 調用發送手機短信驗證碼 from celery_task.task import send_messagephone = 134******** send_message.delay(phone)

總結

以上是生活随笔為你收集整理的python 异步定时任务的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。