django-task 1.4.2

快速启动

安装django任务：

pip install django-task

将其添加到已安装的应用程序中

INSTALLED_APPS=(...'django_rq','django_task',...)

添加django任务的url模式：

urlpatterns=[...url(r'^django_task/',include('django_task.urls',namespace='django_task')),...]

功能

目的

• 以编程方式或从管理员创建异步任务
• 从管理员监视异步任务
• 将所有任务记录到数据库中以备日后检查
• 可选地将特定于任务的日志保存在文本字段和/或文件字段中

详细信息

1. 每个特定的作业都是从models.task派生的模型中描述的。 负责：
   • 在可用队列中选择消费者队列的名称
   • 收集并保存关联作业所需的所有参数
   • 异步运行特定作业
2. 可以运行新作业：
   • 从django管理员创建任务
   • 从代码创建任务，然后调用task.run（）
3. 作业执行工作流：
   • 作业执行由task.run（is_async）触发
   • 作业将接收task.id，并从中检索参数（task.retrieve_params_as_dict（））
   • 在开始时，job会将任务状态更新为"started"，并保存job.id以供参考
   • 在执行过程中，作业可以更新进度指示器
   • 完成后，任务状态最终更新为"成功"或"失败"
   • 有关示例，请参见example.jobs.count\u beans

屏幕截图

应用程序设置

默认值："redis://localhost:6379/0"

运行测试

代码真的有用吗？

source <YOURVIRTUALENV>/bin/activate
(myenv) $ pip install tox
(myenv) $ tox

支持工作类

从0.3.0版开始，添加了一些便利性：

• 不再需要作业函数的@job decorator，因为task.run（）现在 使用queue.enqueue（）而不是jobfunc.delay（），并检索队列 直接从任务本身命名
• 每个任务可以设置自己的任务超时值（以秒为单位）。 当提供时将覆盖默认队列超时
• 提供了一个新的job类，以便在 执行jobfunc之后

classJob(object):@classmethoddefrun(job_class,task_class,task_id):job_trace('job.run() enter')task=Noneresult='SUCCESS'failure_reason=''try:# this raises a "Could not resolve a Redis connection" exception in sync mode#job = get_current_job()job=get_current_job(connection=redis.Redis.from_url(REDIS_URL))# Retrieve task obj and set as Startedtask=task_class.get_task_from_id(task_id)task.set_status(status='STARTED',job_id=job.get_id())# Execute job passing by taskjob_class.execute(job,task)exceptExceptionase:job_trace('ERROR: %s'%str(e))job_trace(traceback.format_exc())iftask:task.log(logging.ERROR,str(e))task.log(logging.ERROR,traceback.format_exc())result='FAILURE'failure_reason=str(e)finally:iftask:task.set_status(status=result,failure_reason=failure_reason)try:job_class.on_complete(job,task)exceptExceptionase:job_trace('NESTED ERROR: Job.on_completed() raises error "%s"'%str(e))job_trace(traceback.format_exc())job_trace('job.run() leave')@staticmethoddefon_complete(job,task):pass@staticmethoddefexecute(job,task):pass

因此，您可以重写 run（） 来实现不同的逻辑， 或者（在大多数情况下）只需提供自己的 execute（） 方法，并且可以选择 覆盖 on_complete（） 以在作业完成后执行清理操作；

示例：

classCountBeansJob(Job):@staticmethoddefexecute(job,task):params=task.retrieve_params_as_dict()num_beans=params['num_beans']foriinrange(0,num_beans):time.sleep(0.01)task.set_progress((i+1)*100/num_beans,step=10)@staticmethoddefon_complete(job,task):print('task "%s" completed with: %s'%(str(task.id),task.status))# An more realistic example from a real project ...# if task.status != 'SUCCESS' or task.error_counter > 0:#    task.alarm = BaseTask.ALARM_STATUS_ALARMED#    task.save(update_fields=['alarm', ])

执行

运行消费者：

python manage.py runserver

运行工作线程：

python manage.py rqworker low high default
python manage.py rqworker low high default
...

示例任务

fromdjango.dbimportmodelsfromdjango.confimportsettingsfromdjango_task.modelsimportTaskclassSendEmailTask(Task):sender=models.CharField(max_length=256,null=False,blank=False)recipients=models.TextField(null=False,blank=False,help_text='put addresses in separate rows')subject=models.CharField(max_length=256,null=False,blank=False)message=models.TextField(null=False,blank=True)TASK_QUEUE=settings.QUEUE_LOWTASK_TIMEOUT=60LOG_TO_FIELD=TrueLOG_TO_FILE=FalseDEFAULT_VERBOSITY=2@staticmethoddefget_jobfunc():from.jobsimportSendEmailJobreturnSendEmailJob

通过重写verbosity属性，可以动态更改 详细信息

当使用 log_to_file=true 时，您可能需要将清理处理程序添加到 删除相应记录时删除日志文件：

pip install django-task

pip install django-task

示例作业

pip install django-task

样本管理命令

pip install django-task

延迟任务检索以避免作业与任务竞争条件

提供了一个helper task.get_task_from_id（）classmethod来检索任务对象 安全地从任务ID。

任务队列创建一种新的竞争条件。为什么？ 因为消息队列很快！ 多快？ 比数据库快。

https://speakerdeck.com/siloraptor/django-tasty salard-dos-donts-using-celeriy

类似的通用助手可用于工作派生需求：

pip install django-task

如何在同一台计算机上为不同实例分隔作业

在同一台机器上分离不同实例的任务（或者更精确地说 对于同一个redis连接），覆盖每个实例的队列名称；

例如：

pip install django-task

然后按如下方式运行Worker:

pip install django-task

如何使用cron计划作业

调用管理命令"count_beans"，它依次执行所需的作业。

例如：

pip install django-task

提供了一个基类taskcommand来简化任何特定 任务相关管理commad；

派生的管理命令只负责：

• 定义适当的命令行参数
• 选择特定的任务类和作业功能

例如：

pip install django-task

javascript帮助程序

提供了一些实用程序视图，用于与来自javascript的任务进行交互。

参考文献：

• 一个为rq（redis queue）提供django集成的简单应用程序
• django中的异步任务与django rq
• django-rq-redux：高级技术和工具
• 基准：共享与专用redis实例
• django美味沙拉-罗伯托·罗萨里奥（roberto rosario）禁止和禁止使用芹菜