【python小随笔】celery异步任务与调用返回值
发布时间:2020-12-20 10:03:53 所属栏目:Python 来源:网络整理
导读:PS:设置完异步任务后,如果出现文件名与依赖包冲突,那一定是你直接运行文件了,最好是在接口里面执行异步任务规避掉这个问题 s1.py(配置任务文件) from celery import Celery timemy_task = Celery( " tasks " ,broker= redis://127.0.0.1:6379 " ) # 为应
PS:设置完异步任务后,如果出现文件名与依赖包冲突,那一定是你直接运行文件了,最好是在接口里面执行异步任务规避掉这个问题 s1.py(配置任务文件) from celery import Celery time my_task = Celery("tasks",broker=redis://127.0.0.1:6379") # 为应用创建任务,func1 @my_task.task(name=Celery.celery.s1.func2) # 指定任务路径(坑) def func2(x,y): time.sleep(3) print(2222222222222222222return x+y ?s2.py(执行异步文件) from celery.result AsyncResult from Celery.celery.s1 func2,my_task time if __name__ == '__main__': 将任务交给Celery的Worker执行 res = func2.delay(2,3 异步获取任务返回值 for i in range(100): time.sleep(1) async_task = AsyncResult(id=res.id,app=my_task) async_task.id,async_task.id) 判断异步任务是否执行成功 if async_task.successful(): 获取异步任务的返回值 result = async_task.get() print(result) 执行成功) break else: 任务还未执行完成") ##### 以下是相关参数 @celery.task(bind=True,name=name) function_name(): pass task方法参数 name:可以显式指定任务的名字;默认是模块的命名空间中本函数的名字。 serializer:指定本任务的序列化的方法; bind:一个bool值,设置是否绑定一个task的实例,如果绑定,task实例会作为参数传递到任务方法中,可以访问task实例的所有的属性,即前面反序列化中那些属性 base:定义任务的基类,可以以此来定义回调函数,默认是Task类,我们也可以定义自己的Task类 default_retry_delay:设置该任务重试的延迟时间,当任务执行失败后,会自动重试,单位是秒,默认3分钟; autoretry_for:设置在特定异常时重试任务,默认False即不重试; retry_backoff:默认False,设置重试时的延迟时间间隔策略; retry_backoff_max:设置最大延迟重试时间,默认10分钟,如果失败则不再重试; retry_jitter:默认True,即引入抖动,避免重试任务集中执行; 当bind=True时,add函数第一个参数是self,指的是task实例 @task(bind=True) 第一个参数是self,使用self.request访问相关的属性 add(self,x,y): try: logger.info(self.request.id) except: self.retry() 当任务失败则进行重试 celery class MyTask(celery.Task): 任务失败时执行 on_failure(self,exc,task_id,args,kwargs,einfo): {0!r} failed: {1!r}.format(task_id,exc)) 任务成功时执行 on_success(self,retval,kwargs): pass 任务重试时执行 on_retry(self,1)">pass @task(base=MyTask) add(x,1)">raise KeyError() 方法相关的参数 exc:失败时的错误的类型; task_id:任务的id; args:任务函数的参数; kwargs:键值对参数; einfo:失败或重试时的异常详细信息; retval:任务成功执行的返回值; 4:TASK的一般属性: ? Task.name:任务名称;
Task.request:当前任务的信息;
Task.max_retries:设置重试的最大次数
Task.throws:预期错误类的可选元组,不应被视为实际错误,而是结果失败;
Task.rate_limit:设置此任务类型的速率限制
Task.time_limit:此任务的硬限时(以秒为单位)。
Task.ignore_result:不存储任务状态。默认False;
Task.store_errors_even_if_ignored:如果True,即使任务配置为忽略结果,也会存储错误。
Task.serializer:标识要使用的默认序列化方法的字符串。
Task.compression:标识要使用的默认压缩方案的字符串。默认为task_compression设置。
Task.backend:指定该任务的结果存储后端用于此任务。
Task.acks_late:如果设置True为此任务的消息将在任务执行后确认 ,而不是在执行任务之前(默认行为),即默认任务执行之前就会发送确认;
Task.track_started:如果True任务在工作人员执行任务时将其状态报告为“已启动”。默认是False;
? 获取任务结果和状态:r = task.apply_async() r.ready() 查看任务状态,返回布尔值,任务执行完成,返回 True,否则返回 False. r.wait() 会阻塞等待任务完成,返回任务执行结果,很少使用; r.get(timeout=1) 获取任务执行结果,可以设置等待时间,如果超时但任务未完成返回None; r.result 任务执行结果,未完成返回None; r.state PENDING,START,SUCCESS,任务当前的状态 r.status 任务成功返回true r.traceback 如果任务抛出了一个异常,可以获取原始的回溯信息 自定义发布者,交换机,路由键,队列,优先级,序列方案和压缩方法: task.apply_async((2,2),compression=zlibjsonpriority.highweb.add0,exchange=web_exchange') (编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |
相关内容
- python – django的Komodo 7或8代码完成
- Python数据包嗅探/数据包捕获 – pcapy不捕获数据包
- 详解python如何调用C/C++底层库与互相传值
- importlib 根据字符串导入模块
- python爬虫-'gbk' codec can't encode ch
- python – 用于汇总下几个项目的数据类型
- python3 pandas 读取MySQL数据和插入的实例
- python-如果金字塔会话是单向散列的,而不是存储在服务器端,
- 如何知道是否发布了Python multiprocessing.Lock?
- 如何更改Squish中waitForObject函数使用的默认超时