Celery简单使用（python开发包使用）

Celery简单使用

• 1：配置文件
• 2：任务文件
• 3：定时文件
• 4：使用流程

1：配置文件

文件名：config.py

import celery
broker = 'redis://localhost:6379/11' backend = 'redis://localhost:6379/12' cel = celery.Celery('tasks',broker=broker, backend=backend) cel.conf.timezone = 'Asia/Shanghai' cel.conf.enable_utc = False 

2：任务文件

文件名：tasks.py

from config import cel import datetime def printLog(data, filename='log.txt'): timestr = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') with open(filename, "a+") as f: f.write("\n+++++++++" + timestr + "++++++++++++\n") f.write(data) f.write("\n++++++++++++++++++++++++++++++++++++++++\n") f.close() @cel.task def add(x,y): printLog(str(x+y)) return x+y

@cel.task def odd(x,y): printLog(str(x + y)) return x-y if __name__ == "__main__": pass 

3：定时文件

文件名：cron.py

from config import cel from celery.schedules import crontab import datetime

cel.conf.update( # task_serializer='json', # accept_content=['json'],  # Ignore other content # result_serializer='json', # timezone='Asia/Shanghai', # enable_utc=False, beat_schedule={ 'task1': { 'task': 'tasks.add', 'schedule': datetime.timedelta(seconds=5), 'args': (3,6), }, # 'task2': { #     'task': 'tasks.odd', #     'schedule': datetime.timedelta(seconds=7), #     'args': (13,6), # }, 'task3': { 'task': 'tasks.odd', 'schedule': crontab(minute="*/1"), 'args': (13,2), }, } ) 

4：使用流程

1：worker 启动：

windows中执行
celery -A tasks worker --pool=solo -l info

Ubuntu中执行
celery -A tasks worker --loglevel=info
celery -A tasks worker -l info

2：beat 启动

定时任务
celery -A cron beat -l info
celery -A cron beat --detach -l info

3：普通异步任务

文件名：test.py

from tasks import add
add.delay(5,8) 

python test.py

本文地址：https://blog.csdn.net/weixin_43853990/article/details/108864812