Python 全栈系列124 redis消息队列搭建-后记_redis:x:124:

说明

之前搭建的时候大概只花了2天，有些地方研究不是很透，这边就对使用后的一些效果做跟踪和改进。

内容

以QA的方式来看：

• Q1: 发现有消息遗漏，怎么回事 ? resp = req.get('http://%s:20000/descrbie_Qs/' % remote_ip)
• A1: 大约调用了30万次,其中有一条消息没有处理，并且一直留在队列中。
  • 怀疑是某一次的删除指令丢失了
  • 从这个角度上，也可以给服务可靠性做一个评估0.9999966666666666,也就是5个9。
  • 另外，这个问题通过其他的方法是可以弥补的
  • 问题的根源来自于固有的丢包（可能也和m4的网络位置有关系，连着交换机不太灵）

通过xrange方法可以获取未删除的消息，一些其他的使用方法参考

# xrange(self, name, min=’-’, max=’+’, count=None):获取消息列表，会自动过滤已经删除的消息

new_q = 'THE_PROBLEM_Q'
res_list = lq_consumer.conn.xrange(new_q)

# 对这些消息可以进行重发再删除
resp = req.post('http://%s:20000/ent_recog/' % remote_ip,json=data)
lq_consumer.del_msg(new_q,'1638430481779-0' )
1
2
3
4
5
6
7
8

• Q2: 在调用002功能时似乎出现了服务器空转的情况，怎么回事？

• A2: 这是由于老程序固有的缺陷和配置不当，导致了在消息快速流入的情况下出现了不满足服务持续运行的状态。

  • 1 老程序是按照允许处理的最大文件数量进行配置。如果当前的的文件集没有被删除就不会继续执行。
  • 2 由于002是由两个功能合并而成的，故障的产生是因为消息流入太快，在识别公司名时耗费了太长时间，导致人名处理的是完全不同的项目，合并流程始终无法照到公司名和人名都识别完成的任务，一直跳过
  • 问题的原因是老程序设计上的缺陷，事先已经知道,如果老程序也改为差集流动就没问题了。
  • 补充：问题的原因是因为数据存在重复，导致有一些任务无法清除。这占用了老程序的最大允许任务份额，导致老程序不处理。

• Q3: 有些结果没有交付，是什么原因？

• A3: 原始数据中存在重复数据，实际上是交付完整了。

  • 对未来设计的改进：增加非重复数据的条数的计算
  • 这个设计的目的是，如果没有完全交付，可以改变状态重新交付，而不是重新计算。

• 1 检查元数据：发现交付数量(按执行前后的id差计算）和原始数据集的数量比较，的确漏了一部分

发现一共有10个文件未交付
In [2]: ls ./sche_001/output
t_sp_court_1638450135297170_TASK_f_002.pkl ...

选择一个元数据
task_meta = fs.from_pickle('t_sp_court_1638450135297170_TASK_f_002', './sche_001/tasks/')

{'task_prefix': 't_sp_court',
 'task_id': '1638450135297170',
 'parts': 1,
 'rec_num': 257,
 'min_rec_id': '1156292',
 'max_rec_id': '4110908',
 'finished_part_num': 1,
 'proc_status': 'proc_007_v_000',
 'is_enable': 1,
 'is_claim': 1,
 'is_done': 1,
 'result_status': 0,
 'is_fetched': 1,
 'is_delivered': 0,
 'is_overdue': 0,
 'ctime': 1638450137.6302016,
 'ttl': 3000,
 'max_retries': 3,
 'funtion_code': '002',
 'delivered_num': 174}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

• 2 检查交付的结果内容

task_data = fs.from_pickle('t_sp_court_1638450135297170_TASK_f_002', './sche_001/output/')

 [174 rows x 2 columns]]
1
2
3

发现结果真的只有174行

• 3 再往前追溯，看原始数据

input_data = fs.from_pickle('t_sp_court_1638450135297170_TASK_f_002', './sche_001/input/')

In [17]: len(input_id_set)
Out[17]: 257

In [19]: len(set(input_id_set))
Out[19]: 174

In [18]: len(post_id_set)
Out[18]: 174
1
2
3
4
5
6
7
8
9
10

结论，原始数据发了重复数据。应该增加非重记录数，交付应该按这个来。

• Q4: IO方面似乎存在BUG，许多任务没有删除。
• A4: 经检查后，确实发现有一些任务没有清除。这是一个bug
  • 1 我不太确定结果是否进行了多次交付，因为delivered_num只是计算插入前后的id差
  • 2 一开始我发现有275个任务没有清除，将is_delivered改为1后，发现还有105没有清除
  • 3 后来我修改了流程的配置参数，将删除过程proc_008_v_001下的模型改为force,过程只要提交了一次数据库，不论如何都删除。将所有文件都清空了。

结论：这次的PM搭建还是有考虑不周的地方,仍然存在bug。但是服务在修补的过程是非常快速而清晰的。

如果业务很重要，是可以在这一版修改的。并且这类型的错误可以学习，并在以后的模式中进行提升（一个错误可以避免未来的无数次错误，经验可以累积）

其他

• 1 修改配置即生效，无需重启服务。这次验证了这一点，的确很方便。例如修改最大允许文件数量，在下一次轮到过程运行时立即生效。
• 2 未来如果对交付的明细要确认的化，可能还需要增加一个delivered_task_id_recs。