python学习笔记3－celery分布式任务处理器

import sys
reload(sys)
sys.setdefaultencoding(‘utf-8’)  # 不加这句话，打印中文log会出错
 
from celery import Celery
 
celery = Celery(‘tasks‘, broker=‘redis://127.0.0.1:6379/0‘) ＃选择本地redis db＝0 作为消息载体, 第一个参数为任务名称
from celery.utils.log import get_task_logger # 倒入celery 中的log模块
logger = get_task_logger(__name__)
 
@celery.task(bind=True, max_retries=10,
             default_retry_delay=1 * 6)  # bind 表示开启, max_retries 是重新尝试的次数,default_retry_delay 是默认的间隔时间，尝试的时间
def exec_task_order_overtime(self, order_id):  # 订单到期后,执行订单失效的任务 
    try:
        logger.info(‘===================> exec_task_order_overtime order_id=%s‘ % order_id)
        success = BaseHandler.context_services.order_overtime_task_service.process_over_time(order_id)
        if success is False:
            logger.error(
                ‘<================order_overtime_task_service.process_over_time Failed, order_id=%s‘ % order_id)
            raise Return(False)
        else:
            logger.info(
                ‘<=================order_overtime_task_service.process_over_time Success, order_id=%s‘ % order_id)
    except Exception as exc:
        logger.info(‘exec_task_order_overtime retry, order_id=%s‘ % order_id)
        raise self.retry(exc=exc, countdown=3)  # 3 秒后继续尝试, 此处的countdown 优先级高于装饰器中的default_retry_delay
 

from celery import Celery
 
celery = Celery(‘tasks‘, broker=‘redis://127.0.0.1:6379/0‘) #消息载体
push_task_id = celery.send_task(‘tasks.exec_task_order_overtime‘
                                , [order_id] # 参数，必须为list，具体可见源码，第三个可以为dict，我们这里没有使用
                                , countdown=10)  #延时多久执行 推送消息