一种基于Kafka实现数据同步的方法与流程

一种基于kafka实现数据同步的方法
技术领域
1.本发明属于基础能力-通用后端技术领域，尤其是涉及一种基于kafka实现数据同步的方法。


背景技术：

2.不同业务系统之间经常需要同步数据，有些对于实时性要求不高的业务通常情况下都是采用离线任务进行数据同步，离线任务存在一定的延时性。而对实时性要求比较高的业务系统并不适合离线任务的形式同步数据，而且这类业务系统通常需要做到尽可能接近实时的同步，同步的数据在各个系统之间应保持一致性，并且不能影响当前业务系统的正常操作。因此本发明基于kafka消息队列，依靠本地消息补偿表，确保消息发送方发送的数据一定被投放到消息队列中。
3.目前系统中的实时数据同步采用http或者webservice的形式，严重依赖其他业务系统，如果其他系统服务不可用，将导致当前业务系统业务异常。特别是需要将数据同步给多个业务系统时，严重耦合，不可控。


技术实现要素：

4.有鉴于此，本发明旨在提出一种基于kafka实现数据同步的方法，数据发送方在执行本地事务时，将需要发送的消息体写入本地消息表，记录同步状态值，同步时间等，并在同一个事务中进行提交。与此同时异步发送kafka消息，在异步ack时将本地消息表中同步状态更改为已同步，利用定时任务定时监控本地消息表中消息同步状态，根据状态判断当前是否继续同步或者告警。
5.为达到上述目的，本发明的技术方案是这样实现的：
6.一种基于kafka实现数据同步的方法，包括以下步骤：
7.s1、创建本地消息表；
8.s2、业务数据发送方a将需要发送的数据，转换为json字符串作为消息体；
9.s3、使用ack模式异步发送kafka消息，在回调中回写本地消息表中同步状态值；
10.s4、业务数据发送方a中设计一个定时器，定时检查一次同步失败的数据，并将其再次投放至kafka消息队列中；
11.s5、如果重复步骤s4设定次数依旧失败，则进行告警进行人工干预。
12.进一步的，在步骤s1中，信息表包括消息id、消息体、首次同步时间、最近一次同步时间、同步状态、失败次数字段。
13.进一步的，在步骤s2中，同步状态为待同步，同当前事务一同提交。
14.进一步的，在步骤s3中，如果发送成功则更改同步状态为同步成功，发送失败则更改同步状态为失败。
15.进一步的，在步骤s4中，使用xxl-job设置定时器。
16.进一步的，本方案公开了一种电子设备，包括处理器以及与处理器通信连接，且用
于存储所述处理器可执行指令的存储器，其特征在于：所述处理器用于执行上述一种基于kafka实现数据同步的方法。
17.进一步的，本方案公开了一种服务器，包括至少一个处理器，以及与所述处理器通信连接的存储器，所述存储器存储有可被所述至少一个处理器执行的指令，所述指令被所述处理器执行，以使所述至少一个处理器执行一种基于kafka实现数据同步的方法。
18.进一步的，本方案公开了一种计算机可读取存储介质，存储有计算机程序，所述计算机程序被处理器执行时实现一种基于kafka实现数据同步的方法。
19.相对于现有技术，本发明所述的一种基于kafka实现数据同步的方法具有以下有益效果：
20.本发明所述的一种基于kafka实现数据同步的方法，尽可能的做到接近实时的同步；数据同步操作解耦，不影响正常业务操作；数据同步失败使用定时任务进行重试，保证数据的最终一致性。
附图说明
21.构成本发明的一部分的附图用来提供对本发明的进一步理解，本发明的示意性实施例及其说明用于解释本发明，并不构成对本发明的不当限定。在附图中：
22.图1为本发明实施例所述的一种基于kafka实现数据同步的方法示意图。
具体实施方式
23.需要说明的是，在不冲突的情况下，本发明中的实施例及实施例中的特征可以相互组合。
24.下面将参考附图并结合实施例来详细说明本发明。
25.一种基于kafka实现数据同步的方法，包括以下步骤：
26.s1、创建本地消息表；
27.s2、业务数据发送方a将需要发送的数据，转换为json字符串作为消息体；
28.s3、使用ack模式异步发送kafka消息，在回调中回写本地消息表中同步状态值；
29.s4、业务数据发送方a中设计一个定时器，定时检查一次同步失败的数据，并将其再次投放至kafka消息队列中；
30.s5、如果重复步骤s4设定次数依旧失败，则进行告警进行人工干预。
31.在步骤s1中，信息表包括消息id、消息体、首次同步时间、最近一次同步时间、同步状态、失败次数字段。
32.在步骤s2中，同步状态为待同步，同当前事务一同提交。
33.在步骤s3中，如果发送成功则更改同步状态为同步成功，发送失败则更改同步状态为失败。
34.在步骤s4中，使用xxl-job设置定时器。
35.本方案公开了一种电子设备，包括处理器以及与处理器通信连接，且用于存储所述处理器可执行指令的存储器，其特征在于：所述处理器用于执行上述一种基于kafka实现数据同步的方法。
36.本方案公开了一种服务器，包括至少一个处理器，以及与所述处理器通信连接的
存储器，所述存储器存储有可被所述至少一个处理器执行的指令，所述指令被所述处理器执行，以使所述至少一个处理器执行一种基于kafka实现数据同步的方法。
37.本方案公开了一种计算机可读取存储介质，存储有计算机程序，所述计算机程序被处理器执行时实现一种基于kafka实现数据同步的方法。
38.本领域普通技术人员可以意识到，结合本文中所公开的实施例描述的各示例的单元及方法步骤，能够以电子硬件、计算机软件或者二者的结合来实现，为了清楚地说明硬件和软件的可互换性，在上述说明中已经按照功能一般性地描述了各示例的组成及步骤。这些功能究竟以硬件还是软件方式来执行，取决于技术方案的特定应用和设计约束条件。专业技术人员可以对每个特定的应用来使用不同方法来实现所描述的功能，但是这种实现不应认为超出本发明的范围。
39.在本技术所提供的几个实施例中，应该理解到，所揭露的方法和系统，可以通过其它的方式实现。例如，以上所述单元的划分，仅仅为一种逻辑功能划分，实际实现时可以有另外的划分方式，例如多个单元或组件可以结合或者可以集成到另一个系统，或一些特征可以忽略，或不执行。上述单元可以是或者也可以不是物理上分开的，作为单元显示的部件可以是或者也可以不是物理单元，即可以位于一个地方，或者也可以分布到多个网络单元上。可以根据实际的需要选择其中的部分或者全部单元来实现本发明实施例方案的目的。
40.最后应说明的是：以上各实施例仅用以说明本发明的技术方案，而非对其限制；尽管参照前述各实施例对本发明进行了详细的说明，本领域的普通技术人员应当理解：其依然可以对前述各实施例所记载的技术方案进行修改，或者对其中部分或者全部技术特征进行等同替换；而这些修改或者替换，并不使相应技术方案的本质脱离本发明各实施例技术方案的范围，其均应涵盖在本发明的权利要求和说明书的范围当中。
41.以上所述仅为本发明的较佳实施例而已，并不用以限制本发明，凡在本发明的精神和原则之内，所作的任何修改、等同替换、改进等，均应包含在本发明的保护范围之内。

技术特征：
1.一种基于kafka实现数据同步的方法，其特征在于，包括以下步骤：s1、创建本地消息表；s2、业务数据发送方a将需要发送的数据，转换为json字符串作为消息体；s3、使用ack模式异步发送kafka消息，在回调中回写本地消息表中同步状态值；s4、业务数据发送方a中设计一个定时器，定时检查一次同步失败的数据，并将其再次投放至kafka消息队列中；s5、如果重复步骤s4设定次数依旧失败，则进行告警进行人工干预。2.根据权利要求1所述的一种基于kafka实现数据同步的方法，其特征在于，在步骤s1中，信息表包括消息id、消息体、首次同步时间、最近一次同步时间、同步状态、失败次数字段。3.根据权利要求1所述的一种基于kafka实现数据同步的方法，其特征在于，在步骤s2中，同步状态为待同步，同当前事务一同提交。4.根据权利要求1所述的一种基于kafka实现数据同步的方法，其特征在于，在步骤s3中，如果发送成功则更改同步状态为同步成功，发送失败则更改同步状态为失败。5.根据权利要求1所述的一种基于kafka实现数据同步的方法，其特征在于，在步骤s4中，使用xxl-job设置定时器。6.一种电子设备，包括处理器以及与处理器通信连接，且用于存储所述处理器可执行指令的存储器，其特征在于：所述处理器用于执行上述权利要求1-5任一所述的一种基于kafka实现数据同步的方法。7.一种服务器，其特征在于：包括至少一个处理器，以及与所述处理器通信连接的存储器，所述存储器存储有可被所述至少一个处理器执行的指令，所述指令被所述处理器执行，以使所述至少一个处理器执行如权利要求1-5任一所述的一种基于kafka实现数据同步的方法。8.一种计算机可读取存储介质，存储有计算机程序，其特征在于：所述计算机程序被处理器执行时实现权利要求1-5任一项所述的一种基于kafka实现数据同步的方法。

技术总结
本发明提供了一种基于Kafka实现数据同步的方法，包括以下步骤：S1、创建本地消息表；S2、业务数据发送方A将需要发送的数据，转换为JSON字符串作为消息体；S3、使用ACK模式异步发送KAFKA消息，在回调中回写本地消息表中同步状态值；S4、业务数据发送方A中设计一个定时器，定时检查一次同步失败的数据，并将其再次投放至KAFKA消息队列中；S5、如果重复步骤S4设定次数依旧失败，则进行告警进行人工干预。本发明有益效果：一种基于Kafka实现数据同步的方法，尽可能的做到接近实时的同步；数据同步操作解耦，不影响正常业务操作；数据同步失败使用定时任务进行重试，保证数据的最终一致性。性。性。


技术研发人员：邹虎
受保护的技术使用者：紫光云技术有限公司
技术研发日：2023.05.11
技术公布日：2023/8/1