celery定时任务

from .. import app
import time

def test11():
    time.sleep(1)
    print(‘test11‘)

def test22():
    time.sleep(2)
    print(‘test22‘)
    test11()

@app.task
def test1_run(): 
　　test11() 
test22() 

from .. import app
import time

def test33():
    time.sleep(3)
    print(‘test33‘)

def test44():
    time.sleep(4)
    print(‘test44‘)
    test33()

@app.task
def test2_run():
    test33()
    test44() 

# 拒绝隐式引入，如果celery.py和celery模块名字一样，避免冲突，需要加上这条语句
# 该代码中，名字是不一样的，最好也要不一样
from __future__ import absolute_import
from celery import Celery

app = Celery(‘tasks‘)
app.config_from_object(‘celery_task.celeryconfig‘) 

from __future__ import absolute_import
from celery.schedules import crontab
from datetime import timedelta

# 使用redis存储任务队列
broker_url = ‘redis://127.0.0.1:6379/7‘
# 使用redis存储结果
result_backend = ‘redis://127.0.0.1:6379/8‘

task_serializer = ‘json‘
result_serializer = ‘json‘
accept_content = [‘json‘]
# 时区设置
timezone = ‘Asia/Shanghai‘
# celery默认开启自己的日志
# False表示不关闭
worker_hijack_root_logger = False
# 存储结果过期时间，过期后自动删除
# 单位为秒
result_expires = 60 * 60 * 24

# 导入任务所在文件
imports = [
    ‘celery_task.app_scripts.test1‘,
    ‘celery_task.app_scripts.test2‘
]

# 需要执行任务的配置
beat_schedule = {
    ‘test1‘: {
        # 具体需要执行的函数
        # 该函数必须要使用@app.task装饰
        ‘task‘: ‘celery_task.app_scripts.test1.test1_run‘,
        # 定时时间
        # 每分钟执行一次，不能为小数
        ‘schedule‘: crontab(minute=‘*/1‘),
        # 或者这么写，每小时执行一次
        # "schedule": crontab(minute=0, hour="*/1")
        # 执行的函数需要的参数
        ‘args‘: ()
    },
    ‘test2‘: {
        ‘task‘: ‘celery_task.app_scripts.test2.test2_run‘,
        # 设置定时的时间，10秒一次
        ‘schedule‘: timedelta(seconds=10),
        ‘args‘: ()
    }
}