diff --git a/script/sql/vincent/V1.10.1_20261013.sql b/script/sql/vincent/V1.10.1_20261013.sql index ea6fe0e3..20fd8dfc 100644 --- a/script/sql/vincent/V1.10.1_20261013.sql +++ b/script/sql/vincent/V1.10.1_20261013.sql @@ -1,46 +1,18 @@ -CREATE TABLE `SZSFJ_ZFJDM_XZJC_INFO` ( - `ID` bigint NOT NULL AUTO_INCREMENT COMMENT '主键,自增ID', - `YWID` varchar(64) NOT NULL COMMENT '业务ID', - `BJQYMC` varchar(128) DEFAULT NULL COMMENT '被检查企业名称', - `BJQYTYSHXYDM` varchar(32) DEFAULT NULL COMMENT '被检查企业统一社会信用代码', - `JCSX` varchar(512) DEFAULT NULL COMMENT '检查事项', - `RQSJ` datetime DEFAULT NULL COMMENT '入企时间', - `ZFYJ` varchar(512) DEFAULT NULL COMMENT '执法依据', - `JCLB` varchar(2) DEFAULT NULL COMMENT '检查类别:1行政检查|2行政处罚|3行政许可|4行政强制|5行政征收|6行政给付|7行政确认|8行政登记|9行政裁决|10文书送达', - `JCYY` varchar(2) DEFAULT NULL COMMENT '检查原因:01日常检查|02投诉举报|03上级交办|04部门移送|05重点领域治理部署|06专项行动|09其它', - `SFLHJC` tinyint DEFAULT NULL COMMENT '是否联合检查:0否,1是', - `LHJCBH` varchar(50) DEFAULT NULL COMMENT '联合检查编号', - `SFWTJC` tinyint DEFAULT NULL COMMENT '是否委托检查:0否,1是', - `WTJCBMMC` varchar(128) DEFAULT NULL COMMENT '委托检查部门名称', - `SFSM` tinyint DEFAULT NULL COMMENT '是否扫码:0是,1否', - `WSMYY` varchar(256) DEFAULT NULL COMMENT '未扫码原因:0其他|1企业原因|2系统原因|3扫码未能显示企业信息|9补录', - `QTYYXXQ` varchar(512) DEFAULT NULL COMMENT '其他原因详情', - `ZFRYXM` varchar(64) DEFAULT NULL COMMENT '执法人员姓名', - `XZZFZH` varchar(64) DEFAULT NULL COMMENT '行政执法证号', - `QTZFRYXM` varchar(64) DEFAULT NULL COMMENT '其他执法人员姓名', - `QTXZZFZH` varchar(64) DEFAULT NULL COMMENT '其他行政执法证号', - `ZFDWMC` varchar(64) DEFAULT NULL COMMENT '执法单位名称', - `ZFBMMC` varchar(64) DEFAULT NULL COMMENT '执法部门名称', - `YWCJSJ` datetime DEFAULT NULL COMMENT '业务创建时间', - `YWXGSJ` datetime DEFAULT NULL COMMENT '业务修改时间', - `CJSJ` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', - `XGSJ` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间', - PRIMARY KEY (`ID`), - UNIQUE KEY `uk_ywid` (`YWID`), - KEY `idx_bjqy_credit` (`BJQYTYSHXYDM`), - KEY `idx_rqsj` (`RQSJ`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='行政检查信息表'; --- 定时任务:每天凌晨2点同步 enforcement_registration 增量数据到 SZSFJ_ZFJDM_XZJC_INFO -INSERT INTO sys_job (job_name, job_group, invoke_target, cron_expression, misfire_policy, concurrent, status, create_by, - create_time, remark) -VALUES ('行政检查信息增量同步', +INSERT INTO sys_job (job_name, job_group, invoke_target, cron_expression, + misfire_policy, concurrent, status, create_by, create_time, remark) +VALUES ('省数据编目-行政检查信息同步到前置库', 'DEFAULT', 'xzjcInfoSyncJob.execute()', '0 0 2 * * ?', - '3', - '1', - '0', - 'admin', - NOW(), + '3', -- misfire: 放弃错过的执行 + '1', -- concurrent: 禁止并发 + '0', -- status: 正常 + 'admin', NOW(), '每天凌晨2点将enforcement_registration增量数据同步到SZSFJ_ZFJDM_XZJC_INFO表(市司法局数据共享)'); + +ALTER TABLE enforcement_registration + MODIFY COLUMN update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间'; + + +update enforcement_registration set update_time = create_time where update_time is null; diff --git a/script/sql/vincent/V1.10.1_20261013_tgk.sql b/script/sql/vincent/V1.10.1_20261013_tgk.sql new file mode 100644 index 00000000..4cd7a9fd --- /dev/null +++ b/script/sql/vincent/V1.10.1_20261013_tgk.sql @@ -0,0 +1,34 @@ +DROP TABLE IF EXISTS `SZSFJ_ZFJDM_XZJC_INFO`; +CREATE TABLE `SZSFJ_ZFJDM_XZJC_INFO` ( + `ID` bigint NOT NULL AUTO_INCREMENT COMMENT '主键,自增ID', + `YWID` varchar(64) NOT NULL COMMENT '业务ID', + `BJQYMC` varchar(128) DEFAULT NULL COMMENT '被检查企业名称', + `BJQYTYSHXYDM` varchar(32) DEFAULT NULL COMMENT '被检查企业统一社会信用代码', + `JCSX` varchar(512) DEFAULT NULL COMMENT '检查事项', + `RQSJ` datetime DEFAULT NULL COMMENT '入企时间', + `ZFYJ` varchar(512) DEFAULT NULL COMMENT '执法依据', + `JCLB` varchar(2) DEFAULT NULL COMMENT '检查类别:1行政检查|2行政处罚|3行政许可|4行政强制|5行政征收|6行政给付|7行政确认|8行政登记|9行政裁决|10文书送达', + `JCYY` varchar(2) DEFAULT NULL COMMENT '检查原因:01日常检查|02投诉举报|03上级交办|04部门移送|05重点领域治理部署|06专项行动|09其它', + `SFLHJC` tinyint DEFAULT NULL COMMENT '是否联合检查:0否,1是', + `LHJCBH` varchar(50) DEFAULT NULL COMMENT '联合检查编号', + `SFWTJC` tinyint DEFAULT NULL COMMENT '是否委托检查:0否,1是', + `WTJCBMMC` varchar(128) DEFAULT NULL COMMENT '委托检查部门名称', + `SFSM` tinyint DEFAULT NULL COMMENT '是否扫码:0是,1否', + `WSMYY` varchar(256) DEFAULT NULL COMMENT '未扫码原因:0其他|1企业原因|2系统原因|3扫码未能显示企业信息|9补录', + `QTYYXXQ` varchar(512) DEFAULT NULL COMMENT '其他原因详情', + `ZFRYXM` varchar(64) DEFAULT NULL COMMENT '执法人员姓名', + `XZZFZH` varchar(64) DEFAULT NULL COMMENT '行政执法证号', + `QTZFRYXM` varchar(64) DEFAULT NULL COMMENT '其他执法人员姓名', + `QTXZZFZH` varchar(64) DEFAULT NULL COMMENT '其他行政执法证号', + `ZFDWMC` varchar(64) DEFAULT NULL COMMENT '执法单位名称', + `ZFBMMC` varchar(64) DEFAULT NULL COMMENT '执法部门名称', + `YWCJSJ` datetime DEFAULT NULL COMMENT '业务创建时间', + `YWXGSJ` datetime DEFAULT NULL COMMENT '业务修改时间', + `CJSJ` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `XGSJ` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间', + `ZLBS` varchar(2) DEFAULT '0' COMMENT '增量标识,0|未同步,1|已同步', + PRIMARY KEY (`ID`), + UNIQUE KEY `uk_ywid` (`YWID`), + KEY `idx_bjqy_credit` (`BJQYTYSHXYDM`), + KEY `idx_rqsj` (`RQSJ`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='行政检查信息表'; diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/mapper/XzjcInfoSyncMapper.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/mapper/XzjcInfoSyncMapper.java index 6769824a..d2621c11 100644 --- a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/mapper/XzjcInfoSyncMapper.java +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/mapper/XzjcInfoSyncMapper.java @@ -1,6 +1,5 @@ package com.zdxt.enforcementcode.mapper; -import com.zdxt.enforcementcode.domain.bo.XzjcInfoSyncBo; import org.apache.ibatis.annotations.Param; import java.util.Date; @@ -8,19 +7,14 @@ import java.util.List; import java.util.Map; /** - * 行政检查信息同步 Mapper 接口 - *
增量同步 enforcement_registration -> SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)
+ * 行政检查信息同步 Mapper 接口(主库) + *仅承载源表 enforcement_registration 的增量分页读取(走默认 master 数据源)
+ *目标表 zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO 的水位查询与写入见 {@link XzjcInfoSyncTgkMapper}
* * @author zdxt */ public interface XzjcInfoSyncMapper { - /** - * 查询增量水位:目标表已同步数据的最大业务修改时间(MAX(YWXGSJ)) - * - * @return 水位时间;目标表为空时返回 null - */ - Date selectWatermark(); /** * 按 (update_time, id) 键集分页查询源表增量数据 @@ -36,11 +30,4 @@ public interface XzjcInfoSyncMapper { @Param("endTime") Date endTime, @Param("pageSize") int pageSize); - /** - * 批量 upsert 目标表(基于 uk_ywid 唯一键 INSERT ... ON DUPLICATE KEY UPDATE,幂等) - * - * @param list 写入行集合 - * @return 影响行数(MySQL upsert 语义,不代表真实写入条数) - */ - int upsertBatch(@Param("list") List增量同步 enforcement_registration -> zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)
+ *仅承载目标表操作(水位查询/批量写入);源表分页读取见 {@link XzjcInfoSyncMapper}
+ * + * @author zdxt + */ +@DS("tgk") +public interface XzjcInfoSyncTgkMapper { + + /** + * 查询增量水位:目标表已同步数据的最大业务修改时间(MAX(YWXGSJ)) + * + * @return 水位时间;目标表为空时返回 null + */ + Date selectWatermark(); + + /** + * 批量 upsert 目标表(基于 uk_ywid 唯一键 INSERT ... ON DUPLICATE KEY UPDATE,幂等) + * + * @param list 写入行集合 + * @return 影响行数(MySQL upsert 语义,不代表真实写入条数) + */ + int upsertBatch(@Param("list") List每天将 enforcement_registration 增量数据同步到 SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)
+ *每天将 enforcement_registration 增量数据同步到 zfjd_tgk 库的 SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)
*默认 cron: 0 0 2 * * ?(每天凌晨 2 点)
* * @author zdxt diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/service/impl/XzjcInfoSyncServiceImpl.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/service/impl/XzjcInfoSyncServiceImpl.java index b1c8ac4c..1548d316 100644 --- a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/service/impl/XzjcInfoSyncServiceImpl.java +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/service/impl/XzjcInfoSyncServiceImpl.java @@ -3,6 +3,7 @@ package com.zdxt.enforcementcode.service.impl; import com.zdxt.enforcementcode.domain.bo.XzjcInfoSyncBo; import com.zdxt.enforcementcode.mapper.XzjcInfoSyncMapper; +import com.zdxt.enforcementcode.mapper.XzjcInfoSyncTgkMapper; import com.zdxt.enforcementcode.service.IXzjcInfoSyncService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -20,13 +21,9 @@ import java.util.Map; import java.util.concurrent.TimeUnit; /** - * 行政检查信息同步服务 - *增量同步 enforcement_registration -> SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)
- *增量水位:目标表 MAX(YWXGSJ),源表 update_time >= 水位即为增量;目标表为空时首次自动全量
- *写入策略:基于 uk_ywid 唯一键 INSERT ... ON DUPLICATE KEY UPDATE(幂等)
- *读取策略:(update_time, id) 键集分页,单页单批,避免一次性加载全量数据导致内存溢出
- *并发控制:Redisson 分布式锁,覆盖 Quartz 调度重叠、手动触发及多实例部署场景
- *SQL 定义见 mapper/enforcementcode/XzjcInfoSyncMapper.xml
+ * 行政检查信息增量同步服务实现 + *源表:master.enforcement_registration;目标表:zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO(推管库,经 @DS("tgk") 数据源写入)
+ *SQL 定义见 mapper/enforcementcode/XzjcInfoSyncMapper.xml 与 XzjcInfoSyncTgkMapper.xml
* * @author zdxt */ @@ -44,8 +41,12 @@ public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService { /** 同步分布式锁 key */ private static final String LOCK_KEY = "xzjc_info_sync:lock"; + /** 源表读取(master) */ private final XzjcInfoSyncMapper xzjcInfoSyncMapper; + /** 目标表写入与水位查询(zfjd_tgk) */ + private final XzjcInfoSyncTgkMapper xzjcInfoSyncTgkMapper; + private final RedissonClient redissonClient; /** @@ -82,7 +83,7 @@ public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService { private int doSync() { // 数据截止时间:今天零点(即只同步前一天晚上 23:59:59 及更早的数据),避免同步当日仍在变化的数据 Date endTime = Timestamp.valueOf(LocalDate.now().atStartOfDay()); - Date watermark = xzjcInfoSyncMapper.selectWatermark(); + Date watermark = xzjcInfoSyncTgkMapper.selectWatermark(); Date cursorTime = watermark; String cursorId = null; int total = 0; @@ -118,7 +119,7 @@ public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService { if (limitReached) { log.warn("[XzjcInfoSync] 达到单次最大分页数 {}(约 {} 条), 剩余数据下次任务继续同步", MAX_PAGES, (long) MAX_PAGES * PAGE_SIZE); } - log.info("[XzjcInfoSync] 增量同步完成: 共处理 {} 条, 起始水位={}, 截止={}", total, watermark, endTime); + log.info("[XzjcInfoSync] 增量同步完成(zfjd_tgk): 共处理 {} 条, 起始水位={}, 截止={}", total, watermark, endTime); return total; } @@ -126,16 +127,24 @@ public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService { * 页级批量 upsert;整批失败时降级为逐条写入,跳过脏数据行,避免单行异常导致游标卡死 */ private void upsertSafely(List