Flask celery非同步任務佇列的使用

celery非同步任務佇列

目錄結構樹：

配置檔案config.py:

# 設定中間人位址

broker_url = 'redis:'

import sys

import os
from celery import celery
from flask import flask
from flask_mail import mail
celery_dir = os.path.dirname(os.getcwd())
sys.path.insert(0, celery_dir)
import config
mail = mail()
# 建立celery物件並設定
# 載入配置
celery.config_from_object('celery_tasks.config')
taskbase = celery.task
class contexttask(taskbase):
abstract = true
def __call__(self, *args, **kwargs):
return taskbase.__call__(self, *args, **kwargs)
celery.task = contexttask
# 啟動celery worker時自動發現任務
celery.autodiscover_tasks(['celery_tasks.email',])
return celery

from flask_mail import message

import config
from celery_tasks.main import celery, mail
# 使用裝飾器將send_email函式裝飾成任務函式
@celery.task(name='send_email')
def send_email(to, subject, html_message):
msg = message(
subject,
sender=config.config.mail_username,
html=html_message,
recipients=[to]
)mail.send(msg)
if __name__ == '__main__':
send_email.delay('[email protected]', 'xx', 'xx')

celery worker -a main.celery -l info

send_email.delay('[email protected]', 'xx', 'xx')

Flask Celery 非同步任務

pip install redis pip install celery 之所以要安裝redis，是因為需要redis作為celery的訊息中介軟體 新建乙個celery fun.py作為celery的配置檔案，並把所有相關方法放進此py。create celery from celery impo...

flask celery 設定定時任務

目前專案用的flask，需要做些定時任務，利用celery，redis,瞎折騰了很久，勉強能實現，做下記錄。專案簡單結構 init pytasks.py其他 from flask import flask from celery import celery 這是flask文件推薦的例項celery應...

非同步 非同步任務

非同步任務 記錄一下學習完springboot後進行使用非同步方法時的筆記 service層 author yesijie date 2019 11 29 15 25 service public class asyncc catch interruptedexception e system.ou...