celery定时任务实践
(window.slotbydup=window.slotbydup || []).push({
id: '2611110',
container: s,
size: '240,200',
display: 'inlay-fix'
您当前位置： &
[ 所属分类
| 时间 2015 |
作者 红领巾 ]
说起celery搞的程序员并不陌生，一般做队列任务之类的总是会用到。最近公司新项目用到类似队列的场景但是还要求定时完成，所以一下想到了celery马上搞起来。
看了资料做了需求分析，celery本身能完成队列和异步任务的功能，但是没有定时功能，只有django-celery里面是有crontab的定时任务功能的。正好我们项目就是用的django所以直接django搞起来
安装不用多说直接pip就搞定, the我们使用的是github上的development版本，与相依赖的是 celery==3.1.18
pip install git+/celery/django-celery.gitpip install celery==3.1.18 # 如果出现依赖问题那就安装
django设置
在settings.py中配置：
import djcelerydjcelery.setup_loader()# BROKER_URL = 'django://' # 直接使用django做broker生产环境不建议，建议使用redis或者rabbitMQBROKER_URL = 'redis://::22222/0' # broker使用reids这是要多2CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' # 定时任务CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend'CELERY_ENABLE_UTC = False # 不是用UTCCELERY_TIMEZONE = 'Asia/Shanghai' CELERY_TASK_RESULT_EXPIRES = 10 #任务结果的时效时间CELERYD_LOG_FILE = BASE_DIR + "/logs/celery/celery.log" # log路径CELERYBEAT_LOG_FILE = BASE_DIR + "/logs/celery/beat.log" # beat log路径CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'msgpack', 'yaml'] # 允许的格式...INSTALLED_APPS = ( ... 'djcelery', 'kombu.transport.django', ...)
第一二项是必须的，在INSTALLED_APPS中添加的djcelery是必须的. kombu.transport.django则是基于Django的broker，如果使用redis就不需要了。
最后创建Celery所需的数据表(django1.8)：
python manage.py migrate
创建一个task
正如前面所说的, 一个task就是一个Pyhton function. 但Celery需要知道这一function是task, 因此我们可以使用celery自带的装饰器decorator: @task. 在django app目录中创建taske.py:
from celery import task@task()def add(x, y): return x + y
当settings.py中的djcelery.setup_loader()运行时, Celery便会查看所有INSTALLED_APPS中app目录中的tasks.py文件, 找到标记为task的function, 并将它们注册为celery task.
将function标注为task并不会妨碍他们的正常执行. 你还是可以像平时那样调用它: z = add(1, 2).
让我们以一个简单的例子作为开始. 例如我们希望在用户发出request后异步执行该task, 马上返回response, 从而不阻塞该request, 使用户有一个流畅的访问过程. 那么, 我们可以使用.delay, 例如在在views.py的一个view中:
from myapp.tasks import add... add.delay(2, 2)...
Celery会将task加入到queue中, 并马上返回. 而在一旁待命的worker看到该task后, 便会按照设定执行它, 并将他从queue中移除. 而worker则会执行以下代码:
import myapp.tasks.addmyapp.tasks.add(2, 2)
启动worker
正如之前说到的, 我们需要worker来执行task. 以下是在开发环境中的如何启动worker:
首先启动terminal, 如同开发django项目一样, 激活virtualenv, 切换到django项目目录. 然后启动django自带web服务器: python manage.py runserver.
然后启动worker:
python manage.py celery worker --loglevel=info
此时, worker将会在该terminal中运行, 并显示输出结果.
打开新的terminal, 激活virtualenv, 并切换到django项目目录:
$ python manage.py shell
&&& from myapp.tasks import add
&&& add.delay(2, 2)
此时, 你可以在worker窗口中看到worker执行该task:
[ 08:47:08,076: INFO/MainProcess] Got task from broker: myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc][ 08:47:08,299: INFO/MainProcess] Task myapp.tasks.add[e080e047-b2a2-43a7-af74-d7d9d98b02fc] succeeded in 0.s: 4
好了，简单的任务我们已经调通，下面我们还是来看定时任务怎么弄。首先新建一个文件task.py（名字可以随便取）
import datetimeimport jsonfrom djcelery import models as celery_modelsfrom django.utils import timezonedef create_task(name, task, task_args, crontab_time): ''' 创建任务 name # 任务名字 task # 执行的任务 "myapp.tasks.add" task_args # 任务参数 {"x":1, "Y":1} crontab_time # 定时任务时间 格式： { 'month_of_year': 9 # 月份 'day_of_month': 5 # 日期 'hour': 01 # 小时 'minute':05 # 分钟 }
''' # task任务， created是否定时创建 task, created = celery_models.PeriodicTask.objects.get_or_create( name=name, task=task) # 获取 crontab crontab = celery_models.CrontabSchedule.objects.filter( **crontab_time).first() if crontab is None: # 如果没有就创建，有的话就继续复用之前的crontab crontab = celery_models.CrontabSchedule.objects.create(**crontab_time) task.crontab = crontab # 设置crontab task.enabled = True # 开启task task.kwargs = json.dumps(task_args) # 传入task参数 expiration = timezone.now() + datetime.timedelta(day=1) task.expires = expiration # 设置任务过期时间为现在时间的一天以后 task.save() return Truedef disable_task(name): ''' 关闭任务 ''' try: task = celery_models.PeriodicTask.objects.get(name=name) task.enabled = False # 设置关闭 task.save() return True except celery_models.PeriodicTask.DoesNotExist: return True
执行定时任务时, Celery会通过celerybeat进程来完成. Celerybeat会保持运行, 一旦到了某一定时任务需要执行时, Celerybeat便将其加入到queue中. 不像worker进程, Celerybeat只有需要一个即可.
python manage.py celery beat --loglevel=info
其实还有一种简单的启动方式worker和beat一起启动：
python manage.py celery worker --loglevel=info --beat
将定期任务储存在django数据库中. 即是在django和celery都运行的状态, 这一方式也可以让我们方便的修改定时任务. 我们只需要设置settings.py中的一项便能开启这一方式:
# settings.pyCELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
然后我们便可以通过django admin的/admin/djcelery/periodictask/添加定期任务了.也可以直接使用刚才我写的脚本在自己的代码逻辑中自己增加和禁用定时任务
可能大家用了一段时间就会想到，很多任务都是一次执行完就不需要，留在数据库里就是垃圾数据了有没有办法清除。方法肯定有因为django-celery本身就有定时任务功能我们加个任务就解决了。好我们看代码：在django app目录中打开taske.py加入如下代码
from djcelery import models as celery_modelsfrom django.utils import timezone@task()def delete(): ''' 删除任务 从models中过滤出过期时间小于现在的时间然后删除 ''' return celery_models.PeriodicTask.objects.filter( expires__lt=timezone.now()).delete()
大家都记得创建任务脚本里设置了 expires 1天以后过期，这样在filter的时候就能当做条件把过期的任务找到并且删除。
然后我们在django-admin中把这个任务加到定时任务中：
名字为del，任务是myapp.taks.delete,定时为每天的5点执行（crontab的格式，不熟悉的大家可以搜索学习下 crontab格式）
本文开发（python）相关术语:python基础教程 python多线程 web开发工程师 软件开发工程师 软件开发流程
转载请注明本文标题：本站链接：
分享请点击：
1.凡CodeSecTeam转载的文章,均出自其它媒体或其他官网介绍,目的在于传递更多的信息,并不代表本站赞同其观点和其真实性负责；
2.转载的文章仅代表原创作者观点,与本站无关。其原创性以及文中陈述文字和内容未经本站证实,本站对该文以及其中全部或者部分内容、文字的真实性、完整性、及时性，不作出任何保证或承若；
3.如本站转载稿涉及版权等问题,请作者及时联系本站,我们会及时处理。
登录后可拥有收藏文章、关注作者等权限...
阅读(5102)
学习的本质是为了解决问题。学以致用 格物致知。
手机客户端
，专注代码审计及安全周边编程，转载请注明出处：http://www.codesec.net
转载文章如有侵权，请邮件 admin[at]codesec.net查看: 5653|回复: 9
UCOS II 系统运行创建了多任务,但是只有一个任务运行了.
主题帖子精华
中级会员, 积分 479, 距离下一级还需 21 积分
在线时间1 小时
UCOS II 系统运行创建了多任务,但是只有一个任务运行了.用的是综合实验的代码修改过来的.
ac7c7eb7a0bc02c2f3d8_572.jpg (0 Bytes, 下载次数: 0)
22:54 上传
运行以后,发现有些功能没执行,然后设置断点时候发现 只有任务watch_task在执行,其他几个任务没执行.不知道是什么问题导致的.各位提个建议我尝试下.
任务里面加了dealy函数没有，如果它的优先级是最高的，又没有dealy，就会出现这样的情况，看看是不是这样的情况，如果不是，就是其他问题了。
ES?全称?Embedded?Show；ES是由众多嵌入式爱好者组织的一个团队，致力于带领学生进步，帮助毕业生找到工作，更重要的是在职的嵌入式工作者提供自主外包和创业的机会。（详情进群查看）
ES的宗旨是：交流共享与合作发展。ES?QQ群?第三群：?。群内论坛有很多嵌入式开发模块解决方案可以参考。进群的立马共享自己做的模块简介。
主题帖子精华
高级会员, 积分 801, 距离下一级还需 199 积分
在线时间256 小时
任务里面加了dealy函数没有，如果它的优先级是最高的，又没有dealy，就会出现这样的情况，看看是不是这样的情况，如果不是，就是其他问题了。
主题帖子精华
金牌会员, 积分 2138, 距离下一级还需 862 积分
在线时间230 小时
可能没加delay()；也可能delay()内的值太小了。还是把这几个任务的代码都贴出来吧。这样子才好分析问题
主题帖子精华
高级会员, 积分 640, 距离下一级还需 360 积分
在线时间27 小时
最好把各个任务的代码粘出来，只看你创建和启动任务太片面了
主题帖子精华
中级会员, 积分 479, 距离下一级还需 21 积分
在线时间1 小时
回复【2楼】TinyBoy:
---------------------------------
还真是...因为最开始是裸写的&.后来加入的ucos&&然后基本用的定时器做标志位调用的函数.,没有用延时的地方尽量没用,所以导致这个优先级最高的任务一直在运行.我这对ucosii还是太不了解了.这基本东西都不知道.
ES?全称?Embedded?Show；ES是由众多嵌入式爱好者组织的一个团队，致力于带领学生进步，帮助毕业生找到工作，更重要的是在职的嵌入式工作者提供自主外包和创业的机会。（详情进群查看）
ES的宗旨是：交流共享与合作发展。ES?QQ群?第三群：?。群内论坛有很多嵌入式开发模块解决方案可以参考。进群的立马共享自己做的模块简介。
主题帖子精华
中级会员, 积分 479, 距离下一级还需 21 积分
在线时间1 小时
回复【4楼】augustedward:
---------------------------------
多谢了.!真是...因为最开始是裸写的&.后来加入的ucos&&然后基本用的定时器做标志位调用的函数.,没有用延时的地方尽量没用,所以导致这个优先级最高的任务一直在运行.我这对ucosii还是太不了解了.这基本东西都不知道.
ES?全称?Embedded?Show；ES是由众多嵌入式爱好者组织的一个团队，致力于带领学生进步，帮助毕业生找到工作，更重要的是在职的嵌入式工作者提供自主外包和创业的机会。（详情进群查看）
ES的宗旨是：交流共享与合作发展。ES?QQ群?第三群：?。群内论坛有很多嵌入式开发模块解决方案可以参考。进群的立马共享自己做的模块简介。
主题帖子精华
高级会员, 积分 801, 距离下一级还需 199 积分
在线时间256 小时
回复【5楼】a4615800:
---------------------------------
你应该用它UCOS的延时函数，具有延时和调度功能的那2个函数。
主题帖子精华
中级会员, 积分 479, 距离下一级还需 21 积分
在线时间1 小时
回复【7楼】TinyBoy:
---------------------------------
void &OSTimeDly (INT32U ticks) 这个吗?
与下面那个效果是不是一样的?
我用的原子的&那种方法.定义了.ucosIi自然就调用的是系统的延时不知道这个是不是.
#ifdef&OS_CRITICAL_METHOD& //如果OS_CRITICAL_METHOD定义了,说明使用ucosII了.
//--------------------------------------------------------------------------------
//nus为要延时的us数.
void&delay_us(u32&nus)
& & &u32&told,tnow,tcnt=0;
& & &u32&reload=SysTick-&LOAD; //LOAD的值 &&&& &
& & &ticks=nus*fac_&
//需要的节拍数 &&
& & &tcnt=0;
& & &OSSchedLock();
//阻止ucos调度，防止打断us延时
& & &told=SysTick-&VAL;&&&&&&&& //刚进入时的计数器值
tnow=SysTick-&VAL;
if(tnow!=told)
if(tnow&told)tcnt+=told-//这里注意一下SYSTICK是一个递减的计数器就可以了.
else&tcnt+=reload-tnow+ &&&&
if(tcnt&=ticks)//时间超过/等于要延迟的时间,则退出.
OSSchedUnlock();
//开启ucos调度&
//----------------------------------------------------------------------------------------------------------------
//nms:要延时的ms数
void&delay_ms(u16&nms)
&&&&if(OSRunning==TRUE)//如果os已经在跑了 &&&&
& &&if(nms&=fac_ms)//延时的时间大于ucos的最少时间周期&
& & & && & OSTimeDly(nms/fac_ms);//ucos延时
& & nms%=fac_
//ucos已经无法提供这么小的延时了,采用普通方式延时&&&&
delay_us((u32)(nms*1000)); //普通方式延时 & & & & &这里有点不理解....上面延时以后出来以后,这里不又延时一次吗 是不是过了&OSTimeDly(nms/fac_ms);这里以后就跳出去了不会执行这个函数了?
ES?全称?Embedded?Show；ES是由众多嵌入式爱好者组织的一个团队，致力于带领学生进步，帮助毕业生找到工作，更重要的是在职的嵌入式工作者提供自主外包和创业的机会。（详情进群查看）
ES的宗旨是：交流共享与合作发展。ES?QQ群?第三群：?。群内论坛有很多嵌入式开发模块解决方案可以参考。进群的立马共享自己做的模块简介。
主题帖子精华
高级会员, 积分 801, 距离下一级还需 199 积分
在线时间256 小时
回复【8楼】a4615800:
---------------------------------
&一般情况下用OSTimeDly()就行了，但是有些情况，比如一个节拍是5ms的，但是你延时的不是节拍的整数倍，比如101ms，你额外要加个1ms，这样的情况才用自己写的delay函数，当然，一个节拍是1ms也行，但是这样UCOS内核开销比较大，建议用5MS的节拍比较合适。
主题帖子精华
高级会员, 积分 743, 距离下一级还需 257 积分
在线时间10 小时
手动进行任务切换&试一下
人生永远追逐着幻光,但谁把幻光看作幻光,谁便沉入无边的苦海
Powered by本帖子已过去太久远了，不再提供回复功能。}