python使用异步任务celery出现异常崩溃时retry重试

前言:

    python下的celery是啥东西大家应该有了解,是一个异步的任务框架 。话说,  我以前写过一个报警平台的项目,也需要任务的扩展成分布式,当时总是觉得 用celery不是那么太靠谱,所以就自己写了一个分布式的任务派发的系统。 

文章源地址,http://xiaorui.cc/?p=254

今个和朋友聊起了分布式爬虫,这哥们说 任务有时候经常的崩溃,但是celery的retry的机制有些意思,最后看了下文档  ,又研究了下retry的参数,然后把自己的一些实战分享给大家。

#xiaorui.cc
@celery.task(bind=True,max_retries=3,default_retry_delay=1 * 6)
def sum(self, num):
    try:
        f = open('plog','a')
        f.write('retry\n')
        f.close()
        num = num + 1
        return num
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

其实最主要就那几个参数,官网写的也很干练,上来就给个例子。  呵呵 ~ 

bind=True 是开启

max_retries 是重新尝试的此时

default_retry_delay 是默认的间隔时间,尝试的时间

下面的代码,大家应该懂的。  就是捕捉异常。  

countdown 也是时间,这个时间优先级是大于上面的default_retry_delay的。 

wKiom1QFykOCq-JIAADh1XYHEAs045.jpg

这个时候我是可以看到,我刚才设置的,碰到异常之后,重新执行三遍的。 


注意下,这个异常是我自己特意抛出去的,不懂的看上面的py。  还有一点是celery 自己会sleep 时间。  我定义了60s 。 

wKioL1QFzLfDAJRpAAyb981Zi7w970.jpg

然后咱们在测试下,重启celery,任务肯定是正常运行的,毕竟是放在队列里面的。启动celery的时候,他也只是从队列里面取任务。 我写入celery的时候,只要保证后端的队列没挂掉就可以了。


redis 127.0.0.1:6379> lrange celery 0 -1
1) "{\"body\": \"gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBFUEaXAgYXEFhXEGVQVjaG9yZHEHTlUJY2FsbGJhY2tzcQhOVQhlcnJiYWNrc3EJTlUHdGFza3NldHEKTlUCaWRxC1UkYTZkMTJkZTMtYjUzOC00ZjMxLWFiNzMtNjExNTQwYjY5NmZkcQxVB3JldHJpZXNxDUsDVQR0YXNrcQ5VCXRhc2tzLnN1bXEPVQl0aW1lbGltaXRxEE5OhnERVQNldGFxElUgMjAxNC0wOS0wMlQxMjoxMjozOC4wNDE4OTYrMDA6MDBxE1UGa3dhcmdzcRR9cRV1Lg==\", \"headers\": {\"redelivered\": true}, \"content-type\": \"application/x-python-serialize\", \"properties\": {\"body_encoding\": \"base64\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"celery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"correlation_id\": \"a6d12de3-b538-4f31-ab73-611540b696fd\", \"reply_to\": \"4459d9e6-2cff-35c9-be5b-45a2d976911e\", \"delivery_tag\": \"bd4480dd-d04a-4401-876b-831b30b55f4e\"}, \"content-encoding\": \"binary\"}"
2) "{\"body\": \"gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBFUEaXAgYXEFhXEGVQVjaG9yZHEHTlUJY2FsbGJhY2tzcQhOVQhlcnJiYWNrc3EJTlUHdGFza3NldHEKTlUCaWRxC1UkNDczMzFhYTgtNzZhOC00N2E1LTg1MGItNzZkYTY0YjY2YzM1cQxVB3JldHJpZXNxDUsBVQR0YXNrcQ5VCXRhc2tzLnN1bXEPVQl0aW1lbGltaXRxEE5OhnERVQNldGFxElUgMjAxNC0wOS0wMlQxMjoxMjo1NS40NjA0MzArMDA6MDBxE1UGa3dhcmdzcRR9cRV1Lg==\", \"headers\": {\"redelivered\": true}, \"content-type\": \"application/x-python-serialize\", \"properties\": {\"body_encoding\": \"base64\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"celery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"correlation_id\": \"47331aa8-76a8-47a5-850b-76da64b66c35\", \"reply_to\": \"4459d9e6-2cff-35c9-be5b-45a2d976911e\", \"delivery_tag\": \"9fa3c120-0bfd-4453-9539-1465e6e820ff\"}, \"content-encoding\": \"binary\"}"
redis 127.0.0.1:6379>


其实我更关注的是崩溃的处理,比如 我们的celery已经做了分布式扩展了。 当一个node已经去到任务,但是突然oom了,sx了。 我原本以为celery 借助rabbotmq的ack机制,来处理这样的情况,但是我的测试结果告诉我,celery的retry机制,只是限于本地玩耍。  其实我们就算不用他的retry装饰器,也可以自己写个for循环,然后过滤下异常罢了。


我现在的做法是,每次获取到任务,做事情之前,先要回调一个接口,然后把我要做的事情push过去,然后把做个标示位,说自己正在干,如果10分钟之后,还没有把你删掉的话,你就再塞入队列中。 

 

当然方法还是有些挫,但是已经线上跑了一段时间,没什么大问题,只是在任务太多的情况下,监控任务的线程貌似多到崩溃。 后期可以改用gevent pool的方式来进行轮训监控事件是否完成。 

如果你是那种平台性质的任务发布,在页面上长时间是loading….的状态,你容易做出分析的。 


但大家没这么倒霉的~  只要异常处理的好。  


个人还是自己开发的靠谱点。 celery可能是太强大了,忽略了一些地方。 


大家觉得文章对你有些作用! 如果想赏钱,可以用微信扫描下面的二维码,感谢!
另外再次标注博客原地址  xiaorui.cc

发表评论

邮箱地址不会被公开。 必填项已用*标注