数据同步方法、装置、设备以及产品与流程

本技术涉及数据处理，尤其涉及数据同步方法、装置、设备以及产品。

背景技术：

1、台账类数据查询系统承担着海量历史业务数据的检索功能，随着企业业务规模扩大，单表数据量常突破亿级，且关联维度复杂，需支持多参数复合查询，为提升查询响应速度，业界普遍采用canal监听数据库变更日志（binlog），经kafka消息队列转发至业务处理模块，最终将数据写入elasticsearch，借助其倒排索引和多字段聚合能力实现高效的多条件复杂查询。

2、然而，该方案在实际应用中暴露出明显缺陷，首先，为保证数据一致性，canal要求按序消费，但不同业务逻辑复杂度各异，固定批次大小易引发慢消费告警，导致消费延迟严重，其次，瞬时产生的数十gb binlog消息涌入队列，消费侧无法区分正常业务与批量更新，顺序消费致使正常消息被长时间阻塞，同时canal与kafka分区数强绑定，无法通过扩容分区提升吞吐，此外，中间件转发异常或消费端逻辑错误均会造成源库与es数据不一致，传统全量重同步成本极高，且缺乏快速定位手段，运维负担沉重。

3、上述内容仅用于辅助理解本技术的技术方案，并不代表承认上述内容是现有技术。

技术实现思路

1、本技术的主要目的在于提供一种数据同步方法、装置、设备以及产品，旨在解决现有数据同步方案中有序消费性能差以及数据一致性管控难的技术问题。

2、为实现上述目的，本技术提出一种数据同步方法，所述数据同步方法包括：

3、在检测到数据库变更事件时，将所述数据库变更事件发送至消息队列；

4、根据业务关联性，通过所述消息队列将所述数据库变更事件分配至目标内部队列；

5、通过所述目标内部队列对所述数据变更事件进行消费，得到消费处理后的变更数据，并将所述变更数据写入目标存储系统；

6、对源数据库以及所述目标存储系统进行数据量比对，得到比对结果，在所述比对结果为存在数据差异的情况下，根据所述数据差异进行同步，得到数据同步结果。

7、在一实施例中，所述根据业务关联性，通过所述消息队列将所述数据库变更事件分配至目标内部队列的步骤包括：

8、提取所述数据库变更事件中的业务关联键，并通过哈希算法对所述业务关联键进行计算，得到哈希值；

9、通过所述哈希值对若干内部队列的总数进行取模运算，得到目标队列序号；

10、将所述数据库变更事件分配至所述目标队列序号对应的目标内部队列中，使得相同业务关联键的事件进入相同内部队列进行顺序消费，不同业务关联键的事件进入不同内部队列并行消费。

11、在一实施例中，所述通过所述目标内部队列对所述数据变更事件进行消费，得到消费处理后的变更数据，并将所述变更数据写入目标存储系统的步骤包括：

12、在所述目标内部队列中批量拉取数据变更事件，并构建消费任务；

13、监测所述消费任务的接口响应时间，得到实时响应时长；

14、基于所述实时响应时长调整拉取事件数量，得到调整后的批次大小；

15、根据所述调整后的批次大小进行批量拉取，得到优化后的消费任务，并执行所述优化后的消费任务，得到变更数据；

16、将所述变更数据写入所述目标存储系统，得到写入结果，并通过偏移量offset管理器记录所述目标内部队列的消费偏移量。

17、在一实施例中，所述通过所述目标内部队列对所述数据变更事件进行消费，得到消费处理后的变更数据，并将所述变更数据写入目标存储系统的步骤之前，所述方法还包括：

18、实时采集所述消息队列中的事件阻塞数量，得到阻塞检测数据；

19、判断所述阻塞检测数据是否超过阻塞阈值；

20、在所述阻塞检测数据超过所述阻塞阈值时，判定为消费异常，并将所述消息队列中的变更事件转发至加速队列，通过所述加速队列进行按序消费。

21、在一实施例中，所述将所述消息队列中的变更事件转发至加速队列的步骤包括：

22、通过滑动时间窗口统计所述消息队列中的事件出现频率，得到事件频次数据；

23、判断所述事件频次数据是否超过突增阈值；

24、若所述事件频次数据超过所述突增阈值，则将所述消息队列中的变更事件标记为突增流量事件，并根据原始分区序号对所述加速队列进行分区数量取余后，将所述突增流量事件转发至所述加速队列进行按序消费；

25、若所述事件频次数据未超过所述突增阈值，则根据业务关联键对所述加速队列的分区数量进行取余后，将对应的变更事件转发至所述加速队列进行按序消费。

26、在一实施例中，所述对源数据库以及所述目标存储系统进行数据量比对，得到比对结果的步骤包括：

27、根据比对触发时机确定比对任务类型，得到任务类型信息；

28、根据所述任务类型信息构建比对任务参数，所述比对任务参数包括租户标识以及时间范围；

29、根据所述租户标识以及时间范围对所述源数据库进行数据查询，得到第一数据量，根据所述租户标识以及时间范围对所述目标存储系统进行数据查询，得到第二数据量；

30、比对所述第一数据量与所述第二数据量，若所述第一数据量与所述第二数据量不一致，则得到比对结果为存在数据差异。

31、在一实施例中，所述在所述比对结果为存在数据差异的情况下，根据所述数据差异进行同步，得到数据同步结果的步骤包括：

32、根据所述数据差异确定目标租户范围，并通过所述目标租户范围提取比对参数；

33、通过二分法对所述比对参数的取值范围进行递归拆分，得到拆分后的子参数范围；

34、调用业务数据对账接口对所述子参数范围进行数据量比对，得到所述数据差异的具体参数信息；

35、根据所述具体参数信息构建数据补偿任务，并执行所述数据补偿任务得到数据同步结果。

36、此外，为实现上述目的，本技术还提出一种数据同步装置，所述数据同步装置包括：

37、发送模块，用于在检测到数据库变更事件时，将所述数据库变更事件发送至消息队列；

38、分配模块，用于根据业务关联性，通过所述消息队列将所述数据库变更事件分配至目标内部队列；

39、消费模块，用于通过所述目标内部队列对所述数据变更事件进行消费，得到消费处理后的变更数据，并将所述变更数据写入目标存储系统；

40、同步模块，用于对源数据库以及所述目标存储系统进行数据量比对，得到比对结果，在所述比对结果为存在数据差异的情况下，根据所述数据差异进行同步，得到数据同步结果。

41、此外，为实现上述目的，本技术还提出一种数据同步设备，所述设备包括：存储器、处理器及存储在所述存储器上并可在所述处理器上运行的计算机程序，所述计算机程序配置为实现如上文所述的数据同步方法的步骤。

42、此外，为实现上述目的，本技术还提出一种存储介质，所述存储介质为计算机可读存储介质，所述存储介质上存储有计算机程序，所述计算机程序被处理器执行时实现如上文所述的数据同步方法的步骤。

43、此外，为实现上述目的，本技术还提供一种计算机程序产品，所述计算机程序产品包括计算机程序，所述计算机程序被处理器执行时实现如上文所述的数据同步方法的步骤。

44、本技术提出的一个或多个技术方案，至少具有以下技术效果：

45、本技术实施例提出的一种数据同步方法、装置、设备以及产品，通过在检测到数据库变更事件时，将所述数据库变更事件发送至消息队列；根据业务关联性，通过所述消息队列将所述数据库变更事件分配至目标内部队列；通过所述目标内部队列对所述数据变更事件进行消费，得到消费处理后的变更数据，并将所述变更数据写入目标存储系统；对源数据库以及所述目标存储系统进行数据量比对，得到比对结果，在所述比对结果为存在数据差异的情况下，根据所述数据差异进行同步，得到数据同步结果。由此，根据业务关联性将事件分配至内部队列以兼顾顺序消费与并行消费，并通过对源数据库与目标存储系统进行数据量比对以自动发现并补偿数据差异，解决了现有数据同步方案中有序消费性能差以及数据一致性管控难的问题，提高了数据同步的效率。