如果沒有這次全量數據對比工具，那么也許這個歷史問題會繼續隱藏著，直到發生線上事故才暴露出來，畢竟人工抽樣驗證發現的概率只有「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可以定位到關鍵詞的索引，然后[:]切片獲取指定內容。

日志解析訂單號+爬蟲獲取申請單+編寫對比校驗邏輯，全量數據對比工具就完成了?？墒?8500單，上萬級別數據，要全部對比完，至少要幾個小時。是時候使用多線程了。

多線程第一步，拆解數據，將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%概率發現不了問題。最穩妥的辦法，還是全量對比，讓每條數據，都經過對比規則的檢驗。