说明
每个创新都会伴随着一系列的改变。
在使用celery进行异步任务后,产生的一个问题恰好也是因为异步产生的。
内容
1 问题描述
我有一个队列 stream1, 对应的worker1需要周期性的获取数据,对输入的数据进行模式识别后分流。worker1我设施为10秒运行一次。然后我就发现输出队列的数据大约是6~7倍于原始数据。
2 分析
在同步执行的状态下,前面 一个任务没有结束,后面的任务即使到了执行时间也会错过。这个在APS任务里是非常明确的。但由于Celery执行的Worker是异步的,这意味着即使前一个任务没有完成,后一个任务还是会如期启动,另开一个线头。
Worker1之前的模式是采用xrange方式获取数据,在处理完成后才将消息删除。
由于模式识别的过程比较复杂,层层过滤,所以单个worker执行的时间超过了60秒。这样在这批消息删除之前,每次启动的worker都取到了相同的数据,处理后也会输出到结果队列。
3 解决办法
理论上,每次worker的取数应该是采用xfetch比较合理,但是对应的,xfetch会因为worker的中断导致消息残留。所以就要有另一些worker来进行残余消息的检测和处理。结果就是 xfetch worker + residual worker配合,显得麻烦。
过去在同步状态下,我就偷懒,只用一个worker进行xrange,这样只有消息被真实消费才会删除。
xfetch是支持多个worker并行的,而xrange则智能支持单个worker。
所以,本次要做的事就是把xfetch + residual 模式搞一下,以后该用什么模式就什么模式。
4 实践
为每个worker提供一种获取残余消息(residual)的办法,每个小时执行一次即可。普通的worker(fetch)一般是秒级,或者分钟级执行的。
当前的QManager是架在RedisAgent服务上封装的对象,这个对象极大简化了平时的操作。不过之前,并没有完全将QManager与RedisAgent的参数对接,采用了较为简单的方式。
本次需要做的是先使用RedisAgent完成对应的任务,然后将QManager进行升级。
构造测试队列
test_list = [{'doc_id':1, 'content':'first'}, {'doc_id':2, 'content':'ss'}]
qm.ensure_group('test.test.test')
qm.parrallel_write_msg('test.test.test', test_list)
{'status': True, 'msg': 'ok,add 2 of 2 messages'}
获取消息
qm.xfetch('test.test.test', count=1)
{'data': [{'_msg_id': '1718984345178-0', 'doc_id': '1', 'content': 'first'}],
'status': True,
'msg': 'ok'}
- 1 判断是否有延误消息
两个关键参数,一个是队列名称,一个是延误时间。如果不写延误时间,就是看所有的延误。
resp = req.post('http://172.17.0.1:24118/get_pending_msg/',json = {'stream_name':'test.test.test' , 'idle_seconds':20}).json()
resp = req.post('http://172.17.0.1:24118/get_pending_msg/',json = {'stream_name':'test.test.test','idle_seconds':None }).json()
{'status': True,
'msg': 'ok',
'data': [['1718984345178-0', 'consumer1', 36675032, 1]]}
延误时间的最大作用是避免获取短时间内超时的任务(如果任务本身就需要很长时间)
如果data字段长度不为0,那么就会有延误消息,获取最小和最大的id即可。
- 2 根据起止id获取数据
delay_data = resp['data']
start_id = delay_data[0][0]
end_id = delay_data[-1][0]
resp = req.post('http://172.17.0.1:24118/xrange/',json = {'stream_name':'test.test.test' ,
'start_id': start_id,
'end_id':end_id}).json()
{'status': True,
'msg': 'ok',
'data': [{'_msg_id': '1718984345178-0', 'doc_id': '1', 'content': 'first'},
{'_msg_id': '1718984345178-1', 'doc_id': '2', 'content': 'ss'}]}
所以相应低,修改QMananger(version1.3)的xrange方法,并增加xpending方法
xrange
...
# 批量获取数据
def xrange(self, stream_name, count = None, start_id = '-' , end_id ='+'):
cur_count = count or self.batch_size
recs_resp = req.post(self.redis_agent_host + 'xrange/',
json ={'connection_hash':self.redis_connection_hash,
'stream_name':stream_name,
'count':cur_count,
'start_id':start_id,
'end_id':end_id
}).json()
return recs_resp
xpending。原来的接口似乎有点小bug:如果队列没有延误,接口查询会失败
...
def xpending(self, stream_name,count = None, idle_seconds = 3600):
cur_count = count or self.batch_size
# 1 确认是否有延误消息:没有延误消息的情况接口会报错
try:
resp = req.post(self.redis_agent_host + 'get_pending_msg/',
json ={'stream_name': stream_name,
'idle_seconds': idle_seconds
}).json()
# 如果没有数据,直接返回(标准格式)
if len(resp['data']) == 0:
print('No Pending')
return resp
except:
return {'status':True, 'msg':'query pending fail', 'data':[]}
# 2 获取被延误的消息
min_id = resp['data'][0][0]
max_id = resp['data'][-1][0]
return self.xrange(stream_name, count = cur_count, start_id = min_id, end_id = max_id)
Note: 我们对正常执行的任务,感知/容忍的周期为分钟;对延误执行(补漏)的任务,感知/容忍的周期为小时。
来看改造后的QM
# xfetch,但是此时已经无数据可取
qm.xfetch('test.test.test' )
{'status': True, 'msg': 'ok', 'data': []}
# xpending 此时有两条延误较长时间的消息
qm.xpending('test.test.test' , idle_seconds=3600)
{'status': True,
'msg': 'ok',
'data': [{'_msg_id': '1718984345178-0', 'doc_id': '1', 'content': 'first'},
{'_msg_id': '1718984345178-1', 'doc_id': '2', 'content': 'ss'}]}
# 用xrange取出,处理
data_list = qm.xpending('test.test.test' , idle_seconds=3600)['data']
[{'_msg_id': '1718984345178-0', 'doc_id': '1', 'content': 'first'},
{'_msg_id': '1718984345178-1', 'doc_id': '2', 'content': 'ss'}]
# 假设处理完,准备删除消息
data_msg_list = qm.extract_msg_id(data_list)
['1718984345178-0', '1718984345178-1']
qm.xdel('test.test.test', data_msg_list)
{'data': 2, 'status': True, 'msg': 'ok'}
# 再次使用xpending
qm.xpending('test.test.test' , idle_seconds=3600)
{'status': True, 'msg': 'no source data', 'data': []}
另外,xpending中,即使是把pending的消息处理掉了,仍然可以读到pending信息,所以每次会调用一下xrange查询一个不存在的区间,稍微有点浪费。不过考虑到这是补救型的操作,一个小时才运行一次,就没有关系了。
标签:Python,resp,worker,全栈,msg,填坑,test,data,id From: https://blog.csdn.net/yukai08008/article/details/139870889