celery使用的一些小坑和技巧(非从无到有的过程)

时间：2021-08-31 09:28:22

纯粹是记录一下自己在刚开始使用的时候遇到的一些坑，以及自己是怎样通过配合redis来解决问题的。
文章分为三个部分，一是怎样跑起来，并且怎样监控相关的队列和任务；二是遇到的几个坑；三是给一些自己配合redis使用的代码示例。

一.celery使用：
　　Ⅰ.把任务中间件服务器跑起来，rabbitmq-server
　　　　跑起来以后，就能在浏览器(http://localhost:15672/#/queues)里面看中间件里面的相关内容了。
　　　　(如果想把这边的某些队列下面的没有跑完的任务丢弃掉的话，进对应的队列，点击delete)
　　Ⅱ.在终端里面对应路径下面把celery的任务队列跑起来：
　　　　eg:celery -A celery_autowork_task worker --loglevel=info -Q auto_work
　　　　注意里面的各个参数的意义
　　Ⅲ.看具体的每个任务的执行情况，是否成功之类的
　　　　1.安装flower,pip3 install flower
　　　　2.用celery把flower跑起来celery flower
　　　　3.http://localhost:5555在这个链接，或者终端里面看见的就能看具体的每个任务的执行情况
二.坑: 
　　Ⅰ.异常场景:并发数设置成4，然后第一次进去队列的时候是用的第一个，retry的时候，居然在4和2里面都触发了任务。也就是关了一个，开了两个。
　　异常原因:在exception的代码段里面加了retry功能， 然后在try的代码段里面也加了try功能，但是查看retry函数本身的定义，发现它的类型就是exception；也就是调用这个本身就是一个异常会走到处理异常的代码段里面去；所以更好的方式是只在exception的代码段里面retry，然后如果try里面需要retry的话，直接用raise 搞一个异常出来；
　　Ⅱ.异常场景:在task装饰器里面定义重试间隔时间default_retry_delay，或者在retry里面指定countdown为5秒，但是在终端里面看见的一个任务的两次开始执行时间发现都会大于5秒。开始还以为是不是两个设置会互相影响。
　　异常原因:后来发现的default_retry_dalay是在retry的时候没有传参数的的默认重试间隔时间，可以看模块代码http://docs.celeryproject.org/en/latest/_modules/celery/app/task.html?highlight=task；但之所以一个任务的两次开始执行时间大于5秒是因为，我们的retry一般都写在exception的代码块里面，而这个间隔时间，其实指的是，执行到retry的时间往后推5秒。try里面的操作会有耗时，所以导致了两次任务的开始时间大于5秒。那么如果想要两次的开始时间为5秒的话，可以在进入任务的时候进行一下计时，然后任务重试的时候看整个try的代码块的耗时。
　　Ⅲ.异常场景:在task装饰器里面提供了max_retries最大重试次数，然后到达最大重试次数(比如10次)以后，在第十次的重试里面仍然会走到retry方法，这个时候，居然在终端开间异常抛出了。
　　异常原因:按我的个人理解，这种设定了10次重试次数，但是在第十次开启第11次的重试时候，应该模块是自动不让重试才对的。可是貌似没有完成，或者可能是我没有搜索到正确的使用方法；解决办法就是自己在重试之前看当前的重试次数，自己控制
　　Ⅳ.异常场景:如果一次任务在第一次的发起，重试次数还没到的时候想要第二次发起怎么办？就是一般来说还是只允许一个实际任务只有一个队列里面的任务处理它
　　处理办法:不管是delay还是apply_async方法都是有返回值的，这个返回值的是celery里面的一个类，可以string以后得到任务的ID，然后之后用app.control里面的revoke方法把对应的任务中断掉。可以查看对应模块的源代码http://docs.celeryproject.org/en/latest/_modules/celery/app/control.html
三.代码示例
　　celery_autowork_task.py【任务执行方法的定义的地方】

from celery import Celery

AutoWork = Celery('auto_work', broker = 【CELERY_BROKER】, backend = '')

AutoWork.conf.CELERY_TIMEZONE              = 'Asia/Shanghai'  # 时区

AutoWork.conf.CELERYD_CONCURRENCY          = 4                # 任务并发数

AutoWork.conf.CELERYD_TASK_SOFT_TIME_LIMIT = 300              # 任务超时时间

AutoWork.conf.CELERY_DISABLE_RATE_LIMITS   = True             # 任务频率限制开关

AutoWork.conf.CELERY_ROUTES = {                               # 任务调度队列

    "autowork_check_barcode_recharge":{"queue":"auto_work"},

}

@AutoWork.task(bind=True,name="autowork_check_barcode_recharge",max_retries=15)

def autowork_check_barcode_recharge(self,recharge_id):

    time_begin=datetime.datetime.now()

    try:

        pass

    except Exception as exc:

        retries=self.request.retries

        if retries<self.max_retries:

            delta_second=(datetime.datetime.now()-time_begin).seconds

            if delta_second<5:

                return self.retry(exc = exc,countdown=5-delta_second)

            else:

                return self.retry(exc = exc,countdown=0)

    finally:

        pass

　　test.py【调用任务以及检查任务的执行情况】

def cancel_pre_celery_task_and_excute_next_task(self,recharge_record):

        recharge_work_guid="recharge_work_guid:%d:%s"%(recharge_record.shop_id,recharge_record.num)

        from handlers.celery_autowork_task import autowork_check_barcode_recharge

        if redis.get(recharge_work_guid):  #上次是否有执行这个实际业务中的任务

            from celery_autowork_task import AutoWork

            AutoWork.control.revoke(redis.get(recharge_work_guid).decode('utf-8'),terminate=True) #如果有执行，则中断

        autowork_guid=autowork_check_barcode_recharge.delay(recharge_record.id) #得到本次的任务GUID

        redis.set(recharge_work_guid,str(autowork_guid)) #保存到redis里面去

        redis.expire(recharge_work_guid,3600)

相关文章