如果没有这次全量数据对比工具，那么也许这个历史问题会继续隐藏着，直到发生线上事故才暴露出来，毕竟人工抽样验证发现的概率只有「5.8%」

(资料图)

背景是发票系统有18500个电子发票订单被财务系统驳回了，驳回原因是财务系统上线了全电发票需求，上线后电子发票枚举被误删，无法处理电子发票。需要我们发票系统对这18500电子发票订单，重新触发提票，让发票能正常开出来。也就是，我们需要刷数。刷数是个高危操作，极易引发线上问题。

经验教训告诉我们，刷数虽然是一种处理线上问题的方法，但是也特别容易引起二次事故。对于刷数，我们需要像新需求一样对待，经过完备的需求分析、设计评审、代码评审。主要考虑以下3点：

刷数范围，怎么筛选问题数据，评审过滤条件；刷数程序，怎么修复问题数据，评审代码逻辑；验证方法，怎么验证修复数据，分析测试场景；

在刷数实施前，群里报备，周知相关方及相关人员。先试刷，验证无问题后，再全刷，最后验证问题数据已经得以修复。这一套刷数流程，通过加强前期评审，能很好地预防缺陷，增强刷数成功的信心。但是人工评估可能遗漏场景，可能对真实数据情况把握不全，刷数的关键还在于最后的数据验证。

抽样验证还是全量验证，这是一个问题。抽样验证是人工随机挑几个数据进行验证，我们通常倾向于使用抽样验证，一是抽样验证是一种科学的有效的验证方法，虽然它存在一定概率的遗漏，但是很多时候是可以接受的风险；二是抽样验证也是无奈之举，找不到办法进行全量验证。我们遇到的困难是，数据存在ES中，批量把所有数据查出来很麻烦，也无法直接编写校验逻辑，全量验证似乎是不可能的。

全量验证有2个思路：

如果能直连库，那么先提数，再写程序对比；如果只能WEB页面查数，那么使用Python爬虫提数，再写程序对比；

后者适用于我们的情况。ES能通过WEB页面查询数据，只要是WEB页面，即使有Cookie，也能爬取到接口数据。F12抓包到查询接口的URL、Cookie、入参后，使用Python的requests库可以爬取查询结果数据：

url = "http://xxx"headers = {    "Cookie": "xxx",    "Content-Type": "application/json"}response = request("get", url=url, headers=headers)result = response.json()

我们使用这种方式，传入订单号，查询到了申请单数据，以便进行对比校验逻辑。而订单号，研发打在了日志里，需要下载日志文件后，进行解析，可以使用Python切片：

orders = set()with open("some.log") as f:    for line in f.read().splitlines():        if "xxx" in line:            orders.add(line[line.index("orderId="):line.index(",已执行")].replace("orderId=", ""))print(len(orders))with open("orders.txt", "w") as f:    f.writelines(",".join(orders))

.index可以定位到关键词的索引，然后[:]切片获取指定内容。

日志解析订单号+爬虫获取申请单+编写对比校验逻辑，全量数据对比工具就完成了。可是18500单，上万级别数据，要全部对比完，至少要几个小时。是时候使用多线程了。

多线程第一步，拆解数据，将18500单拆成以100单为一组的列表：

def split_list(lst, size):    """    将列表 lst 拆分成每份 size 个元素的子列表，并返回一个包含所有子列表的列表。    """    return [lst[i:i + size] for i in range(0, len(lst), size)]

多线程第二步，队列提数，让多个线程依次从列表中取出数据，每个线程每次取不同的数据：

import threading# 待处理的数据列表data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]# 创建锁对象lock = threading.Lock()# 定义线程函数def process_data():    global data    while True:        # 加锁        lock.acquire()        # 如果列表为空，说明所有数据已被处理完毕，退出循环        if len(data) == 0:            lock.release()            break        # 取出列表中的第一个数据        num = data.pop(0)        # 释放锁        lock.release()        # 对数据进行处理        print("Processing data:", num)# 创建多个线程threads = []for i in range(5):    t = threading.Thread(target=process_data)    threads.append(t)# 启动所有线程for t in threads:    t.start()# 等待所有线程结束for t in threads:    t.join()print("All data processed.")

阻塞队列是通过加锁来实现的，每个线程在取数前先加锁，然后pop(0)取出列表中的第一个数据，再释放锁。上述程序修改①data②数据处理逻辑③线程数即可使用。

对比工具使用多线程后，运行时间从小时级别降到了分钟级别。当天研发本来以为要跑很久，准备第二天再来看，就先撤了。我执着了一下多线程实现，在ChatGPT帮助下，很快就把结果跑出来。赶紧打电话摇人，让研发回来看问题，研发那时刚到家，掏出钥匙把门打开。在全量对比前，我们也都做了一轮抽样验证，均没有发现任何问题。18500单全量对比，发现有1064单存在问题，能抽样发现的概率只有5.8%。

总结，分析这些问题原因：

遗漏了1种数据情况，评估不到位未考虑到刷数环境影响，在预发环境刷数，上下游环境都是预发，可能跟线上版本不一样，尤其是做写操作时，格外需要注意刷数程序本身缺陷，这个缺陷隐藏在一段用了很多次刷数的历史代码里面，不是100%会导致问题

可以发现，大数据量验证，人工无法百分百保证数据准确性，抽样检查，94.2%概率发现不了问题。最稳妥的办法，还是全量对比，让每条数据，都经过对比规则的检验。