Django中使用Celery执行异步和定时任务的注意事项

WEB前端开发社区昨天

1. Windows中使用Celery 4.0及以后版本

Celery 4.0+及以后版本不支持在windows系统上运行。如果你希望在windows系统上使用celery, 有两种方法。

方法一：安装3.1.25版本

pip install celery==3.1.25

方法二：安装gevent

pip install gevent# 启动workercelery -A <module> worker -l info -P gevent

2. @task与@shared_task的区别

当我们使用@app.task装饰器定义我们的异步任务时，那么这个任务依赖于根据项目名myproject生成的Celery实例。

app = Celery('myproject')
@app.task(bind=True)def debug_task(self):    print('Request: {0!r}'.format(self.request))

然而我们在进行Django开发时为了保证每个app的可重用性，我们经常会在每个app文件夹下编写异步任务，这些任务并不依赖于具体的Django项目名。使用@shared_task 装饰器能让我们避免对某个项目名对应Celery实例的依赖，使app的可移植性更强。

from __future__ import absolute_importfrom celery import shared_task
@shared_taskdef add(x, y):    return x + y

3. 如果异步的任务包括耗时的I/O操作

一个无限期阻塞的任务会使得工作单元无法再做其他事情。如果你的任务里有 I/O 操作，请确保给这些操作加上超时时间，例如使用 requests 库时给网络请求添加一个超时时间：

connect_timeout, read_timeout = 5.0, 30.0response = requests.get(URL, timeout=(connect_timeout, read_timeout))

默认的 prefork 池调度器对长时间任务不是很友好，所以如果你的任务需要运行很长时间，确保在启动工作单元时使能了 -ofair 选项。

4. 使用多装饰器

当使用多个装饰器装饰任务函数时，确保 task 装饰器最后应用（在python中，这意味它必须在第一个位置）：

@app.task@decorator2@decorator1def add(x, y):    return x + y

5. 使用bind=True绑定任务

一个绑定任务意味着任务函数的第一个参数总是任务实例本身(self)，就像 Python 绑定方法类似，如下例所示：

from celery.utils.log import get_task_loggerlogger = get_task_logger(__name__)
@task(bind=True)def add(self, x, y):    logger.info(self.request.id)

绑定任务在这些情况下是必须的：任务重试（使用 app.Task.retry() )，访问当前任务请求的信息，以及你添加到自定义任务基类的附加功能。

6. 忽略不想要的结果

如果你不在意任务的返回结果，可以设置 ignore_result 选项，因为存储结果耗费时间和资源。你还可以可以通过 task_ignore_result 设置全局忽略任务结果。

@app.task(ignore_result=True)def mytask():    something()

7. 避免启动同步子任务

让一个任务等待另外一个任务的返回结果是很低效的，并且如果工作单元池被耗尽的话这将会导致死锁。

# 坏例子

@app.taskdef update_page_info(url):    page = fetch_page.delay(url).get()    info = parse_page.delay(url, page).get()    store_page_info.delay(url, info)
@app.taskdef fetch_page(url):    return myhttplib.get(url)
@app.taskdef parse_page(url, page):    return myparser.parse_document(page)
@app.taskdef store_page_info(url, info):    return PageInfo.objects.create(url, info)

# 好例子

def update_page_info(url):    # fetch_page -> parse_page -> store_page    chain = fetch_page.s(url) | parse_page.s() | store_page_info.s(url)    chain()
@app.task()def fetch_page(url):    return myhttplib.get(url)
@app.task()def parse_page(page):    return myparser.parse_document(page)
@app.task(ignore_result=True)def store_page_info(info, url):    PageInfo.objects.create(url=url, info=info)

在好例子里，我们将不同的任务签名链接起来创建一个任务链，三个子任务按顺序执行。警告：不建议同步执行子任务！

8. Django的模型对象不应该作为参数传递给任务

Django 的模型对象。他们不应该作为参数传递给任务。几乎总是在任务运行时从数据库获取对象是最好的，因为老的数据会导致竞态条件。假象有这样一个场景，你有一篇文章，以及自动展开文章中缩写的任务：

class Article(models.Model):    title = models.CharField()    body = models.TextField()
@app.taskdef expand_abbreviations(article):    article.body.replace('MyCorp', 'My Corporation')    article.save()

首先，作者创建一篇文章并保存，这时作者点击一个按钮初始化一个缩写展开任务：

>>> article = Article.objects.get(id=102)>>> expand_abbreviations.delay(article)

现在，队列非常忙，所以任务在2分钟内都不会运行。与此同时，另一个作者修改了这篇文章，当这个任务最终运行，因为老版本的文章作为参数传递给了这个任务，所以这篇文章会回滚到老的版本。修复这个竞态条件很简单，只要参数传递文章的 id 即可，此时可以在任务中重新获取这篇文章：

@app.taskdef expand_abbreviations(article_id):    article = Article.objects.get(id=article_id)    article.body.replace('MyCorp', 'My Corporation')    article.save()

9. 使用on_commit函数处理事务

我们再看另外一个celery中处理事务的例子。这是在数据库中创建一个文章对象的 Django 视图，此时传递主键给任务。它使用 commit_on_success 装饰器，当视图返回时该事务会被提交，当视图抛出异常时会进行回滚。

from django.db import transaction
@transaction.commit_on_successdef create_article(request):    article = Article.objects.create()    expand_abbreviations.delay(article.pk)

如果在事务提交之前任务已经开始执行会产生一个竞态条件；数据库对象还不存在。解决方案是使用 on_commit 回调函数来在所有事务提交成功后启动任务。

from django.db.transaction import on_commitdef create_article(request):    article = Article.objects.create()    on_commit(lambda: expand_abbreviations.delay(article.pk))

10. 自定义重试延迟

当任务发送例外时，app.Task.retry() 函数可以用来重新执行任务。当一个任务被重试，它在重试前会等待给定的时间，并且默认的由 default_retry_delay 属性定义。默认设置为 3 分钟。注意延迟设置的单位是秒（int 或者 float）。你可以通过提供 countdown 参数覆盖这个默认值。

# retry in 30 minutes.@app.task(bind=True, default_retry_delay=30 * 60)  def add(self, x, y):try:        something_raising()except Exception as exc:# overrides the default delay to retry after 1 minuteraise self.retry(exc=exc, countdown=60)

声明：

本文于网络整理，版权归原作者所有，如来源信息有误或侵犯权益，请联系我们删除或授权事宜。

本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。

WEB前端开发社区 昨天

WEB前端开发社区昨天