[發明專利]一種增量數據實時同步的方法及相關設備

在審

• 申請號: 202011284996.5
• 申請日: 2020-11-17
• 公開（公告）號: CN112487087A
• 公開（公告）日: 2021-03-12
• 發明（設計）人: 宋洋;安靖;胡瀟涵;劉佳;王毅;崔貝貝
• 申請（專利權）人: 中國人壽保險股份有限公司
• 主分類號: G06F16/27
• 分類號: G06F16/27;G06F16/25
• 代理公司: 北京風雅頌專利代理有限公司 11403
• 代理人: 李翔

1.一種增量數據實時同步的方法，其特征在于，包括：

從配置文件中讀取需要同步的MongoDB數據集合，創建數據的同步線程；

所述同步線程從oplog中讀取對應的所述MongoDB數據集合的數據，將讀取到的所述MongoDB數據集合的數據轉換成json格式；

所述同步線程將所述json格式的數據加入到發送列表中并異步發送到與所述MongoDB數據集合對應的Kafka的主題中，同時將異步發送立刻返回的返回值加入到發送列表中；

判斷所述發送列表的數據量是否達到預設的閾值，如果未達到預設的閾值，則繼續發送所述json格式的數據，如果達到閾值，則使用所述Kafka提供的接口判斷是否所述發送列表中的所有數據都發送成功，如果所述發送列表中的所有數據都發送成功，則返回上一步驟，繼續發送下一個批次的所述json格式的數據，如果所述發送列表中的數據存在一個或多個發送失敗，則對所述發送列表中的所有數據進行重復發送，直至發送成功，返回上一步驟，繼續發送下一個批次的所述json格式的數據。

2.根據權利要求1所述的方法，其特征在于，所述同步線程從oplog中讀取對應的所述MongoDB數據集合的數據，還包括所述同步線程從Redis讀取同步時間標記，根據所述同步時間標記從oplog中讀取對應的所述MongoDB數據集合的數據；

所述同步線程所有數據都發送成功之后，將該批次最后一條數據的時間標記記錄到所述Redis中，以實現斷點續傳。

3.根據權利要求2所述的方法，其特征在于，所述根據所述同步時間標記從oplog中讀取對應的所述MongoDB數據集合的數據，包括：

如果所述Redis中有對應的所述MongoDB數據集合的時間標記，則從所述oplog中讀取所述時間標記之后的所述MongoDB數據集合的數據，如果所述Redis中沒有對應的所述MongoDB數據集合的時間標記，則從所述oplog中讀取最新產生的所述MongoDB數據集合的數據。

4.根據權利要求1所述的方法，其特征在于，所述創建數據的同步線程的數量不少于一個；當同步線程多于一個時，每個線程分別執行所述創建數據的同步線程之后的操作。

5.一種增量數據實時同步的裝置，其特征在于，包括：

獲取模塊，被配置為從配置文件中讀取需要同步的MongoDB數據集合，創建數據的同步線程；

轉換模塊，被配置為所述同步線程從oplog中讀取對應的所述MongoDB數據集合的數據，將讀取到的所述MongoDB數據集合的數據轉換成json格式；

發送模塊，被配置為將所述同步線程將所述json格式的數據加入到發送列表中并異步發送到與所述MongoDB數據集合對應的Kafka的主題中，同時將異步發送立刻返回的返回值加入到發送列表中；

判斷模塊，被配置為判斷所述發送列表的數據量是否達到預設的閾值，如果未達到預設的閾值，則繼續發送所述json格式的數據，如果達到閾值，則使用所述Kafka提供的接口判斷是否所述發送列表中的所有數據都發送成功，如果所述發送列表中的所有數據都發送成功，則返回上一步驟，繼續發送下一個批次的所述json格式的數據，如果所述發送列表中的數據存在一個或多個發送失敗，則對所述發送列表中的所有數據進行重復發送，直至發送成功，返回上一步驟，繼續發送下一個批次的所述json格式的數據。

6.根據權利要求5所述的裝置，其特征在于，所述同步線程從oplog中讀取對應的所述MongoDB數據集合的數據，具體被配置為所述同步線程從Redis讀取同步時間標記，根據所述同步時間標記從oplog中讀取對應的所述MongoDB數據集合的數據；

所述判斷模塊之后還包括斷點續傳模塊，被配置為當所述同步線程所有數據都發送成功之后，將該批次最后一條數據的時間標記記錄到所述Redis中，以實現斷點續傳。