标签：自动 项目目录 传递参数 模块 需要 default pat 执行 文件
1、celery执行定时任务
1)几点几分执行：
celery_task-s1.py
from celery import Celery import time # redis不加密码 broker=‘redis://127.0.0.1:6379/0‘ backend=‘redis://127.0.0.1:6379/1‘ app=Celery(‘test‘,backend=backend,broker=broker) @app.task # @app名字.task def add(x,y): time.sleep(2) return x+y
add_task.py
# 2020/2/21 22:38:30开始执行任务 from datetime import datetime v1 = datetime(2020, 3, 21, 22, 38, 30) print(v1) v2 = datetime.utcfromtimestamp(v1.timestamp()) print(v2) # 取出要执行任务的时间对象，调用apply_async方法，args是任务函数的参数，eta是执行的时间 result = celery_task_s1.add.apply_async(args=[1, 3], eta=v2) print(result.id)
此时启动work和broker，程序会等到指定的那个时间才会执行；
2)多长时间后执行
add_task.py
from datetime import datetime ctime = datetime.now() # 默认用utc时间 utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) from datetime import timedelta #取10秒之后的时间对象 time_delay = timedelta(seconds=10) task_time = utc_ctime + time_delay # 使用apply_async并设定时间 result = celery_task_s1.add.apply_async(args=[4, 3], eta=task_time) print(result.id)
此时启动work和broker，程序会等到10秒后才会执行；
2、执行计划任务
类似于contab的定时任务
案例, (用的是多任务机构)：
celery.py
from celery import Celery from datetime import timedelta from celery.schedules import crontab backend=‘redis://127.0.0.1:6379/0‘ broker=‘redis://127.0.0.1:6379/1‘ app=Celery(‘test‘,broker=broker, backend=backend, # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类 include=[‘celery_task.order_task‘, ‘celery_task.user_task‘ ]) # 字典里可以写多个不同的任务 app.conf.beat_schedule = { # 名字随意命名 ‘add-every-10-seconds‘: { # 执行order_task下的test_celery函数 ‘task‘: ‘celery_task.order_task.order_add‘, # 每隔10秒执行一次 # ‘schedule‘: 1.0, # ‘schedule‘: crontab(minute="*/1"), ‘schedule‘: timedelta(seconds=10), # 给任务函数传递参数 ‘args‘: (5, 6) }, # ‘add-every-12-seconds‘: { # ‘task‘: ‘celery_task.order_task.order_add‘, # # 每年3月21号，23点15分执行 # ‘schedule‘: crontab(minute=15, hour=23, day_of_month=21, month_of_year=3), # ‘args‘: (16, 16) # }, }
order_task.py
import time from celery_task.celery import app @app.task def order_add(x, y): time.sleep(1) return x + y
此时需要先启动一个beat：celery beat -A celery_task -l info
前面都是手动提交任务，这个beat会自动帮我们提交任务，而且周期是我们任务里指定的，我这里是10秒，它就会10秒提交一次；
然后启动work，celery worker -A celery_task -l info -P eventlet
work就会每隔10秒接收一次任务，并一直循环执行；
3、django中使用celery
直接把多任务结构copy过来，使用即可；
celery_task是在项目目录下；
celery.py
from celery import Celery backend=‘redis://127.0.0.1:6379/0‘ broker=‘redis://127.0.0.1:6379/1‘ app=Celery(‘test‘,broker=broker, backend=backend, # 包含以下两个任务文件，去相应的py文件中找任务，对多个任务做分类 include=[‘celery_task.order_task‘, ‘celery_task.user_task‘ ])
user_task.py
import time from celery_task.celery import app @app.task def user_add(x, y): time.sleep(1) return x + y
views.py
from django.shortcuts import render, HttpResponse from celery_task.user_task import user_add from celery.result import AsyncResult from celery_task.celery import app # Create your views here. def index(request): result = user_add.delay(8, 9) return HttpResponse(result.id) def check_result(request): res = request.GET.get(‘id‘) async = AsyncResult(id=res, app=app) if async.successful(): # 取出return值 result = async.get() print(result) return HttpResponse(‘任务执行完成‘)
urls.py
from django.conf.urls import url from django.contrib import admin from app01 import views urlpatterns = [ url(r‘^admin/‘, admin.site.urls), url(r‘^index/‘, views.index), url(r‘^check/‘, views.check_result), ]
需要先启动work： celery worker -A celery_task -l info -P eventlet
然后启动django，在浏览器中访问：
先访问了index，index视图函数中执行了任务；
然后拿到任务id，再访问check手动查询结果；
注意：
在celery的任务函数中不能直接调用django的环境，比如要使用ORM就不行，需要手动添加django环境才行；
import os os.environ.setdefault("DJANGO_SETTINGS_MODULE", "BMS.settings") import django # 加载django django.setup() # 启动django from app01 import models # 这样就能使用ORM了 ......
4、django-celery
其实除了原生的celery模块，django也提供了一个django-celery模块，但据说不太好用；
如果使用到了再看吧；
标签：自动 项目目录 传递参数 模块 需要 default pat 执行 文件
原文地址：https://www.cnblogs.com/weiyiming007/p/12543554.html
