Python 全栈系列256 异步任务与队列消息控制(填坑)

说明

每个创新都会伴随着一系列的改变。

在使用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查询一个不存在的区间，稍微有点浪费。不过考虑到这是补救型的操作，一个小时才运行一次，就没有关系了。