[Java知识库] 基于SpringBoot集群数据同步功能设计实现

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> Java知识库 -> 基于SpringBoot集群数据同步功能设计实现 -> 正文阅读

[Java知识库]基于SpringBoot集群数据同步功能设计实现

1、概述

????????基于Web集群数据同步功能设计方案，主题功能主要包括三部分：同步配置、同步集群、同步记录，通过进行同步配置集群地址、请求超时时间、同步表的配置、同步频率信息管控同步任务。其次通过添加集群实现多集群数据同步，最后通过同步记录实时查看同步的结果。

????????摘要：数据同步；集群；Web；

2、功能架构图

????????如图为同步服务功能的架构图，服务器A、B、C之间是完全独立，用户通过集群配置添加服务器地址信息实现多主机之间的数据同步功能。我们以A、B两台服务器举例讲解同步数据流程，第一步，在同步集群中添加服务器B的IP等信息，第二步，配置并开启服务器A同步任务，让A不断向B服务发送同步数据请求，请求到的数据我们将进行数据校验、比对等操作后对其本地数据进行插入或者修改。

3、数据库表设计

?同步集群信息表：主要包括集群地址、频率、在线状态、报警状态等信息。

-- ----------------------------
-- Table structure for sys_cluster_info
-- ----------------------------
DROP TABLE IF EXISTS `sys_cluster_info`;
CREATE TABLE `sys_cluster_info`  (
  `id` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主键',
  `ipaddr` varchar(15) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL COMMENT '集群ip地址',
  `frequency` bigint NULL DEFAULT NULL COMMENT '主机同步频率（单位：秒）',
  `online_status` bit(1) NULL DEFAULT NULL COMMENT '主机在线状态（1:在线 0:不在线）',
  `alarm_status` bit(1) NULL DEFAULT NULL COMMENT '主机同步报警状态（1:正常 0:异常）',
  `create_by` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `create_time` datetime NULL DEFAULT NULL,
  `deleted` bit(1) NULL DEFAULT NULL,
  `enabled` bit(1) NULL DEFAULT NULL,
  `remark` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `sort` bigint NULL DEFAULT NULL,
  `update_by` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `update_time` datetime NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = DYNAMIC;

同步记录表信息：主要包括同步记录时间、同步类型、同步结果状态等信息。

-- ----------------------------
-- Table structure for sys_sync_record
-- ----------------------------
DROP TABLE IF EXISTS `sys_sync_record`;
CREATE TABLE `sys_sync_record`  (
  `id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '主键',
  `time` datetime NULL DEFAULT NULL COMMENT '同步时间',
  `type` bit(1) NULL DEFAULT NULL COMMENT '同步类型（1：表 0：其他）',
  `result_status` bit(1) NULL DEFAULT NULL COMMENT '同步结果状态（1：成功 0：失败）',
  `create_by` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `create_time` datetime NULL DEFAULT NULL,
  `deleted` bit(1) NULL DEFAULT NULL,
  `enabled` bit(1) NULL DEFAULT NULL,
  `remark` text CHARACTER SET utf8 COLLATE utf8_general_ci NULL,
  `sort` bigint NULL DEFAULT NULL,
  `update_by` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `update_time` datetime NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = DYNAMIC;

?4、集群信息

????????通过添加服务器的集群信息实现多主机的同步服务，定时任务会轮询所有在线设备将其数据拉取到本地进行数据同步。

5、同步配置

????????同步配置主要是定时任务的配置参数信息，主要包括：请求地址、请求超时时间、定时任务频率、同表配置（table1，table2），需要主要的是同步表配置多表配置我是以逗号分隔，且表的配置需要考虑顺序，实际同步中要按照表的顺序进行数据同步，以避免业务发生问题。

6、同步数据接口格式

????????服务器同步数据接口是HTTP/GET请求，例：http://IP:8000/api/sync/tables?time=''，同步返回结果如下，time是上次定时任务成功记录的时间，将其发送给服务以避免拉取已同步的数据提高效率，code表示获取同步数据结果状态，msg表示失败信息，data是要拉取的同步表的数据信息，结构如下：

?{
?? ?"code": 0,
?? ?"msg": "",
?? ?"date": "2022-07-19 16:21:18",
?? ?"version": "1.0.0",
?? ?"data": {
?? ??? ?"tableA": [
?? ??? ??? ?{
?? ??? ??? ??? ?"name": "tableA",
?? ??? ??? ??? ?"uuid": "40288b977ffd8599017ffd87243e0001",
?? ??? ??? ??? ?"values": "xxxxx",
?? ??? ??? ??? ?"updateTime": "2022-06-27 10:56:51.0"
?? ??? ??? ?},
?? ??? ??? ?{
?? ??? ??? ??? ?"name": "tableA",
?? ??? ??? ??? ?"uuid": "40288b977ffd8599017ffd87243e0002",
?? ??? ??? ??? ?"values": "xxxxx",
?? ??? ??? ??? ?"updateTime": "2022-06-28 10:56:51.0"
?? ??? ??? ?}
?? ??? ?],
?? ??? ?"tableB": [
?? ??? ??? ?{
?? ??? ??? ??? ?"name": "tableB",
?? ??? ??? ??? ?"uuid": "40288a0480db57ae0180db5973a90003",
?? ??? ??? ??? ?"values": "xxxxx",
?? ??? ??? ??? ?"updateTime": "2022-05-19 16:05:14.0"
?? ??? ??? ?}
?? ??? ?]
?? ?}
}

7、同步数据定时任务

????????定时任务按照频率设置，定时轮询本服务配置的集群，拉取所有IP集群的数据信息，同时在进行同步过程中要判断服务在线状态网络是否可达，在确保集群在线下拉取数据，拉取后根据更新时间、主键判断数据是否是最新数据，然后更新或添加到本地数据库中，并将此次同步记录保存到同步记录表中。

8、通用数据库接口实现

????????定时任务在每次轮询拉取不同服务器的数据格式是json，程序需要将符合要求的不同表的数据更新或插入本地数据库中，所以需要按照将其json数据格式进行解析。主要是需要实现一个SQL原生的通用方法类实现增删改查指定表的数据，由于本项目中使用的是JPA，需要使用EntityManager 来实现原生sql执行，部分案例如下所示。

package com.bamboocloud.modules.sync.repository.common;

import com.bamboocloud.exception.BadRequestException;
import org.springframework.stereotype.Repository;

import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.persistence.Query;
import javax.transaction.Transactional;
import java.util.List;


/**
 * @author Lw
 * @date 2022/7/7 18:55
 * @description
 */
@Repository
public class CommonRepository {

    @PersistenceContext
    private EntityManager entityManager;

    // 查询通过表
    public List findAllByTableAndCol(String name, String cols) {
        String sql = String.format("select %s from %s", cols, name);
        Query query = entityManager.createNativeQuery(sql);
        List data = query.getResultList();
        return data;
    }

    // 查询通过表、时间、列名
    public List findAllByTime(String name, String cols, String time) {
        String sql = String.format("select %s from %s where update_time > ?", cols, name);
        Query query = entityManager.createNativeQuery(sql);
        query.setParameter(1, time);
        List data = query.getResultList();
        return data;
    }


    // 插入
    @Transactional
    public boolean addByTableName(String name, String values) {
        String sql = String.format("insert into %s values(%s)",name,values);
        Query query = entityManager.createNativeQuery(sql);
        int res = query.executeUpdate();
        return res > 0 ? true : false;
    }

    // 修改
    @Transactional
    public void updateByTable(String name, String UUId, String values) {
        // 获得并且开启事物
        try {
            String delSql = String.format("delete from %s where id = ?", name);
            Query delQuery = entityManager.createNativeQuery(delSql);
            delQuery.setParameter(1, UUId);
            int del = delQuery.executeUpdate();
            String addSql = String.format("insert into %s values(%s)", name, values);
            Query addQuery = entityManager.createNativeQuery(addSql);
            int add = addQuery.executeUpdate();
            //提交事务
        } catch (Exception e) {
            throw new BadRequestException("更新数据异常：" + e.getMessage());
        }

    }
}