#SORA#celery研究笔记

标签：

最近看到celery文档task部分，做一下小结

1. 实际处理时，我们可以使用一个类似于logging的模块生成日志。

2. 对于某些任务，你可以设置当满足某些条件时，重试任务、拒绝任务或忽略任务

3. 在定义task时，@app.task(bind=True)中的bind参数可以让你在task中访问请求中的内容，比如id，group之类的信息

4. @app.task(AGRS=VALUE)，ARGS有好几个参数可以设置，比如name，有些和全局设置（CELERY_xxx_xxx之类的）是一样的配置内容

5. 可以自定义任务状态（默认有pending，started，success，failure，retry，revoked）

6. 当你使用pickle作为序列化工具时，你应该定义那些可以被pickle的异常（我用json，直接忽略）

实例化。你可以继承Task类定义新类，定义的__init__方法只会被调用一次，此后将持续存在。当你的task以此新类为基类，后面对此task的调用中，__init__的作用还会存在。（用途：自定义类中维持一个对数据库的连接，task可以不用每次都创建连接，而是对那个存在的属性进行操作）。

e.g：

from celery import Task

class DatabaseTask(Task):
    abstract = True
    _db = None

    @property
    def db(self):
        if self._db is None:
            self._db = Database.connect()
        return self._db

@app.task(base=DatabaseTask)
def process_rows():
    for row in process_rows.db.table.all():
        …

7.定义新类时，设置为抽象类，可以作为task的基类。其中又有四种handle方法（after_return，on_failure，on_retry，on_success）

e.g：

from celery import Task

class DebugTask(Task):
    abstract = True

    def after_return(self, *args, **kwargs):
        print(‘Task returned: {0!r}‘.format(self.request)


@app.task(base=DebugTask)
def add(x, y):
    return x + y

8.最佳实践：

*忽略不需要的结果，或者设置CELERY_TASK_RESULT_EXPIRES

*若不需要，关闭rate limits

*避免写出同步的task（阻塞），同时执行的同步的task，前面的会令后面的低效，应该写成异步的方法或使用callback 

e.g：

#Bad:

@app.task
def update_page_info(url):
    page = fetch_page.delay(url).get()
    info = parse_page.delay(url, page).get()
    store_page_info.delay(url, info)

@app.task
def fetch_page(url):
    return myhttplib.get(url)

@app.task
def parse_page(url, page):
    return myparser.parse_document(page)

@app.task
def store_page_info(url, info):
    return PageInfo.objects.create(url, info)
    
    
    
    
    
#Good:

def update_page_info(url):
    # fetch_page -> parse_page -> store_page
    chain = fetch_page.s() | 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)

*使用细粒度的任务，而不是又长又臭的long task

*尽量把需要操作的数据放在本地，或使worker靠近数据所在，降低因IO延迟的影响

暂时如此，以后有需要再补

#SORA#celery研究笔记

标签：

原文地址：http://my.oschina.net/hochikong/blog/399219