目前專案用的flask，需要做些定時任務，利用celery，redis,瞎折騰了很久，勉強能實現，做下記錄。專案簡單結構
app_

__init__.py

app.py

tasks.py

其他

app.py或者__init__.py ,我是把以下內容寫入了__init__.pyfrom flask import Flask
from celery import Celery
app=Flask(__name__)

#這是flask文件推薦的例項celery應用的函式，相當於重寫了Task的呼叫方法，使其可以應用在flask的上下文中
def make_celery(app):

celery=Celery(app.import_name,broker=app.config['CELERY_BROKER_URL'])

celery.conf.update(app.config)

Taskbase=celery.Task

class ContextTask(Taskbase):

abstract=True

def __call__(self,*args,**kw):

with app.app_context():

return Taskbase.__call__(self,*args,**kw)

celery.Task=ContextTask

return celery

需要新增一些配置
app.config.update(

CELERY_BROKER_URL='redis://localhost:6379',#Broker 地址

CELERY_RESULT_BACKEND='redis://localhost:6379',#

結果儲存地址

#定時任務，

CELERYBEAT_SCHEDULE = {

'task1':{

'task':'app.tasks.test_task',

"schedule": timedelta(seconds=1),

"args":'',

},

'task2': {

'task': 'app.tasks.test_task1',

"schedule": timedelta(seconds=3),

"args": '',

},

}

)
在tasks.py 中可以寫入自己要定時執行的函式
from . import celery
@celery.task
def test_task():

pass

@celery.task
def test_task1():

pass

啟動 celerycelery -A app.tasks:celery worker -l info  -B