Compare commits
6 Commits
master
...
dev-V1.10.1
| Author | SHA1 | Date |
|---|---|---|
vincent
|
0d5b475f45 | 5 days ago |
|
vincent
|
192247cdf6 | 2 weeks ago |
|
vincent
|
08d5d6eb69 | 2 weeks ago |
|
vincent
|
603cf92a66 | 2 weeks ago |
|
vincent
|
fefa6c28e5 | 3 weeks ago |
|
vincent
|
84dd936b17 | 3 weeks ago |
16 changed files with 834 additions and 53 deletions
@ -0,0 +1,18 @@ |
|||
|
|||
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', -- 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; |
|||
@ -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='行政检查信息表'; |
|||
@ -0,0 +1,83 @@ |
|||
package com.zdxt.enforcementcode.domain.bo; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.util.Date; |
|||
|
|||
/** |
|||
* 行政检查信息同步写入行(对应市司法局数据共享表 SZSFJ_ZFJDM_XZJC_INFO) |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@Data |
|||
public class XzjcInfoSyncBo { |
|||
|
|||
/** 业务ID(YWID) */ |
|||
private String ywid; |
|||
|
|||
/** 被检查企业名称(BJQYMC) */ |
|||
private String bjqymc; |
|||
|
|||
/** 被检查企业统一社会信用代码(BJQYTYSHXYDM) */ |
|||
private String bjqytyshxydm; |
|||
|
|||
/** 检查事项(JCSX) */ |
|||
private String jcsx; |
|||
|
|||
/** 入企时间(RQSJ) */ |
|||
private Date rqsj; |
|||
|
|||
/** 执法依据(ZFYJ) */ |
|||
private String zfyj; |
|||
|
|||
/** 检查类别(JCLB) */ |
|||
private String jclb; |
|||
|
|||
/** 检查原因(JCYY) */ |
|||
private String jcyy; |
|||
|
|||
/** 是否联合检查:0否,1是(SFLHJC) */ |
|||
private Integer sflhjc; |
|||
|
|||
/** 联合检查编号(LHJCBH) */ |
|||
private String lhjcbh; |
|||
|
|||
/** 是否委托检查:0否,1是(SFWTJC) */ |
|||
private Integer sfwtjc; |
|||
|
|||
/** 委托检查部门名称(WTJCBMMC) */ |
|||
private String wtjcbmmc; |
|||
|
|||
/** 是否扫码:0是,1否(SFSM) */ |
|||
private Integer sfsm; |
|||
|
|||
/** 未扫码原因:0其他|1企业原因|2系统原因|3扫码未能显示企业信息|9补录(WSMYY) */ |
|||
private String wsmyy; |
|||
|
|||
/** 其他原因详情(QTYYXXQ) */ |
|||
private String qtyyxxq; |
|||
|
|||
/** 执法人员姓名(ZFRYXM) */ |
|||
private String zfryxm; |
|||
|
|||
/** 行政执法证号(XZZFZH) */ |
|||
private String xzzfzh; |
|||
|
|||
/** 其他执法人员姓名(QTZFRYXM) */ |
|||
private String qtzfryxm; |
|||
|
|||
/** 其他行政执法证号(QTXZZFZH) */ |
|||
private String qtxzzfzh; |
|||
|
|||
/** 执法单位名称(ZFDWMC) */ |
|||
private String zfdwmc; |
|||
|
|||
/** 执法部门名称(ZFBMMC) */ |
|||
private String zfbmmc; |
|||
|
|||
/** 业务创建时间(YWCJSJ) */ |
|||
private Date ywcjsj; |
|||
|
|||
/** 业务修改时间(YWXGSJ) */ |
|||
private Date ywxgsj; |
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
package com.zdxt.enforcementcode.mapper; |
|||
|
|||
import org.apache.ibatis.annotations.Param; |
|||
|
|||
import java.util.Date; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
/** |
|||
* 行政检查信息同步 Mapper 接口(主库) |
|||
* <p>仅承载源表 enforcement_registration 的增量分页读取(走默认 master 数据源)</p> |
|||
* <p>目标表 zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO 的水位查询与写入见 {@link XzjcInfoSyncTgkMapper}</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
public interface XzjcInfoSyncMapper { |
|||
|
|||
|
|||
/** |
|||
* 按 (update_time, id) 键集分页查询源表增量数据 |
|||
* |
|||
* @param cursorTime 游标时间(null 时全量) |
|||
* @param cursorId 游标 ID(cursorTime 非空且本参数为 null 时,按 update_time >= cursorTime 含边界防漏) |
|||
* @param endTime 截止时间(上界,不含),只同步该时间之前的数据(通常为今天零点,即前一天晚上的数据) |
|||
* @param pageSize 分页大小 |
|||
* @return 增量数据行(列为 enforcement_registration 小写列名) |
|||
*/ |
|||
List<Map<String, Object>> selectIncrementalPage(@Param("cursorTime") Date cursorTime, |
|||
@Param("cursorId") String cursorId, |
|||
@Param("endTime") Date endTime, |
|||
@Param("pageSize") int pageSize); |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
package com.zdxt.enforcementcode.mapper; |
|||
|
|||
import com.baomidou.dynamic.datasource.annotation.DS; |
|||
import com.zdxt.enforcementcode.domain.bo.XzjcInfoSyncBo; |
|||
import org.apache.ibatis.annotations.Param; |
|||
|
|||
import java.util.Date; |
|||
import java.util.List; |
|||
|
|||
/** |
|||
* 行政检查信息同步 Mapper 接口(zfjd_tgk 推管库) |
|||
* <p>增量同步 enforcement_registration -> zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)</p> |
|||
* <p>仅承载目标表操作(水位查询/批量写入);源表分页读取见 {@link XzjcInfoSyncMapper}</p> |
|||
* |
|||
* @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<XzjcInfoSyncBo> list); |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
package com.zdxt.enforcementcode.quartz.job; |
|||
|
|||
import com.zdxt.enforcementcode.service.IXzjcInfoSyncService; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
/** |
|||
* 行政检查信息同步定时任务 |
|||
* <p>每天将 enforcement_registration 增量数据同步到 zfjd_tgk 库的 SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)</p> |
|||
* <p>默认 cron: 0 0 2 * * ?(每天凌晨 2 点)</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@Slf4j |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class XzjcInfoSyncJob { |
|||
|
|||
private final IXzjcInfoSyncService xzjcInfoSyncService; |
|||
|
|||
/** |
|||
* 任务执行入口(public,供 JobInvokeUtil 通过反射调用) |
|||
*/ |
|||
public void execute() { |
|||
log.info("[XzjcInfoSyncJob] 定时任务触发:行政检查信息增量同步"); |
|||
try { |
|||
xzjcInfoSyncService.syncIncremental(); |
|||
} catch (Exception e) { |
|||
log.error("[XzjcInfoSyncJob] 行政检查信息同步失败", e); |
|||
throw e; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,6 @@ |
|||
package com.zdxt.enforcementcode.service; |
|||
|
|||
public interface IXzjcInfoSyncService { |
|||
|
|||
public int syncIncremental(); |
|||
} |
|||
@ -0,0 +1,243 @@ |
|||
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; |
|||
import org.redisson.api.RLock; |
|||
import org.redisson.api.RedissonClient; |
|||
import org.springframework.stereotype.Service; |
|||
|
|||
import java.sql.Timestamp; |
|||
import java.time.LocalDate; |
|||
import java.time.LocalDateTime; |
|||
import java.util.ArrayList; |
|||
import java.util.Date; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* 行政检查信息增量同步服务实现 |
|||
* <p>源表:master.enforcement_registration;目标表:zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO(推管库,经 @DS("tgk") 数据源写入)</p> |
|||
* <p>SQL 定义见 mapper/enforcementcode/XzjcInfoSyncMapper.xml 与 XzjcInfoSyncTgkMapper.xml</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@Slf4j |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService { |
|||
|
|||
/** 分页读取大小(与批量写入大小一致,一页一批) */ |
|||
private static final int PAGE_SIZE = 500; |
|||
|
|||
/** 单次任务最大分页数,防止异常场景下扫描失控(500*1000=单次最多约 50 万条,剩余下次任务继续) */ |
|||
private static final int MAX_PAGES = 1000; |
|||
|
|||
/** 同步分布式锁 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; |
|||
|
|||
/** |
|||
* 执行一次增量同步 |
|||
* <p>通过 Redis 分布式锁保证同一时刻仅一个任务实例在执行(含多实例部署场景)</p> |
|||
* |
|||
* @return 处理条数;未获取到锁(上一次仍在执行)或被中断时返回 0 |
|||
*/ |
|||
public int syncIncremental() { |
|||
RLock lock = redissonClient.getLock(LOCK_KEY); |
|||
boolean locked = false; |
|||
try { |
|||
// 等待 0 秒:抢不到立即返回,不阻塞 Quartz 线程;leaseTime=-1 启用看门狗自动续期,避免长任务锁提前过期
|
|||
locked = lock.tryLock(0, -1, TimeUnit.SECONDS); |
|||
if (!locked) { |
|||
log.warn("[XzjcInfoSync] 上一次同步尚未结束(锁被占用),本次触发跳过"); |
|||
return 0; |
|||
} |
|||
return doSync(); |
|||
} catch (InterruptedException e) { |
|||
Thread.currentThread().interrupt(); |
|||
log.error("[XzjcInfoSync] 同步任务获取锁时被中断", e); |
|||
return 0; |
|||
} finally { |
|||
if (locked && lock.isHeldByCurrentThread()) { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 分页增量同步主体 |
|||
*/ |
|||
private int doSync() { |
|||
// 数据截止时间:今天零点(即只同步前一天晚上 23:59:59 及更早的数据),避免同步当日仍在变化的数据
|
|||
Date endTime = Timestamp.valueOf(LocalDate.now().atStartOfDay()); |
|||
Date watermark = xzjcInfoSyncTgkMapper.selectWatermark(); |
|||
Date cursorTime = watermark; |
|||
String cursorId = null; |
|||
int total = 0; |
|||
boolean limitReached = false; |
|||
|
|||
for (int page = 1; page <= MAX_PAGES; page++) { |
|||
List<Map<String, Object>> rows = xzjcInfoSyncMapper.selectIncrementalPage(cursorTime, cursorId, endTime, PAGE_SIZE); |
|||
if (rows.isEmpty()) { |
|||
break; |
|||
} |
|||
|
|||
List<XzjcInfoSyncBo> batch = new ArrayList<>(rows.size()); |
|||
for (Map<String, Object> row : rows) { |
|||
batch.add(toBo(row)); |
|||
} |
|||
upsertSafely(batch); |
|||
total += rows.size(); |
|||
|
|||
// 游标推进到本页最后一条,下页从其后继续
|
|||
Map<String, Object> last = rows.get(rows.size() - 1); |
|||
cursorTime = toTimestamp(last.get("update_time")); |
|||
cursorId = str(last.get("id")); |
|||
log.info("[XzjcInfoSync] 第 {} 页完成, 累计 {} 条, 游标=({}, {})", page, total, cursorTime, cursorId); |
|||
|
|||
if (rows.size() < PAGE_SIZE) { |
|||
break; |
|||
} |
|||
if (page == MAX_PAGES) { |
|||
limitReached = true; |
|||
} |
|||
} |
|||
|
|||
if (limitReached) { |
|||
log.warn("[XzjcInfoSync] 达到单次最大分页数 {}(约 {} 条), 剩余数据下次任务继续同步", MAX_PAGES, (long) MAX_PAGES * PAGE_SIZE); |
|||
} |
|||
log.info("[XzjcInfoSync] 增量同步完成(zfjd_tgk): 共处理 {} 条, 起始水位={}, 截止={}", total, watermark, endTime); |
|||
return total; |
|||
} |
|||
|
|||
/** |
|||
* 页级批量 upsert;整批失败时降级为逐条写入,跳过脏数据行,避免单行异常导致游标卡死 |
|||
*/ |
|||
private void upsertSafely(List<XzjcInfoSyncBo> batch) { |
|||
// try {
|
|||
// xzjcInfoSyncTgkMapper.upsertBatch(batch);
|
|||
// } catch (Exception e) {
|
|||
// log.error("[XzjcInfoSync] 批量写入失败,降级逐条写入排查脏数据, 批次大小={}", batch.size(), e);
|
|||
// for (XzjcInfoSyncBo bo : batch) {
|
|||
// try {
|
|||
// xzjcInfoSyncTgkMapper.upsertBatch(List.of(bo));
|
|||
// } catch (Exception rowEx) {
|
|||
// log.error("[XzjcInfoSync] 脏数据跳过: ywid={}", bo.getYwid(), rowEx);
|
|||
// }
|
|||
// }
|
|||
// }
|
|||
|
|||
for (XzjcInfoSyncBo bo : batch) { |
|||
try { |
|||
xzjcInfoSyncTgkMapper.upsertBatch(List.of(bo)); |
|||
} catch (Exception rowEx) { |
|||
log.error("[XzjcInfoSync] 脏数据跳过: ywid={}", bo.getYwid(), rowEx); |
|||
} |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 源表行 -> 目标表写入行(字段映射与长度截断) |
|||
*/ |
|||
private XzjcInfoSyncBo toBo(Map<String, Object> row) { |
|||
Long hasUnion = toLong(row.get("has_union")); |
|||
Long hasAuto = toLong(row.get("has_auto")); |
|||
String unionNo = str(row.get("union_no")); |
|||
// 联合检查判定:has_union=1 且 union_no 不为空
|
|||
boolean isUnion = hasUnion != null && hasUnion == 1 && unionNo != null && !unionNo.isEmpty(); |
|||
|
|||
XzjcInfoSyncBo bo = new XzjcInfoSyncBo(); |
|||
bo.setYwid(str(row.get("id"))); |
|||
bo.setBjqymc(fit(row.get("enterprise_name"), 128)); |
|||
bo.setBjqytyshxydm(fit(row.get("enterprise_number"), 32)); |
|||
bo.setJcsx(fit(coalesce(row, "law_enforcement_item_text", "law_enforcement_item"), 512)); |
|||
bo.setRqsj(toDate(row.get("enforcement_time"))); |
|||
bo.setZfyj(fit(coalesce(row, "law_enforcement_basis_text", "law_enforcement_basis"), 512)); |
|||
bo.setJclb(fit(row.get("law_enforce_category"), 2)); |
|||
bo.setJcyy(fit(row.get("check_category"), 2)); |
|||
bo.setSflhjc(isUnion ? 1 : 0); |
|||
bo.setLhjcbh(fit(unionNo, 50)); |
|||
bo.setSfwtjc(hasAuto != null && hasAuto == 1 ? 1 : 0); |
|||
bo.setWtjcbmmc(fit(row.get("entrust_dept_name"), 128)); |
|||
bo.setSfsm(toLong(row.get("enforce")) == null ? null : toLong(row.get("enforce")).intValue()); |
|||
bo.setWsmyy(fit(row.get("check_css"), 256)); |
|||
bo.setQtyyxxq(fit(row.get("other_reasons"), 512)); |
|||
bo.setZfryxm(fit(row.get("law_enforcer"), 64)); |
|||
bo.setXzzfzh(fit(row.get("law_enforcer_id"), 64)); |
|||
bo.setQtzfryxm(fit(row.get("law_enforcer_two"), 64)); |
|||
bo.setQtxzzfzh(fit(row.get("law_enforcer_id_two"), 64)); |
|||
bo.setZfdwmc(fit(row.get("bureau_dept_name"), 64)); |
|||
bo.setZfbmmc(fit(row.get("law_enforcement"), 64)); |
|||
bo.setYwcjsj(toDate(row.get("create_time"))); |
|||
bo.setYwxgsj(toDate(row.get("update_time"))); |
|||
return bo; |
|||
} |
|||
|
|||
private Object coalesce(Map<String, Object> row, String first, String second) { |
|||
Object v = row.get(first); |
|||
return v != null ? v : row.get(second); |
|||
} |
|||
|
|||
private String str(Object v) { |
|||
return v == null ? null : v.toString(); |
|||
} |
|||
|
|||
private String fit(Object v, int maxLen) { |
|||
String s = str(v); |
|||
if (s != null && s.length() > maxLen) { |
|||
return s.substring(0, maxLen); |
|||
} |
|||
return s; |
|||
} |
|||
|
|||
private Long toLong(Object v) { |
|||
if (v == null) { |
|||
return null; |
|||
} |
|||
if (v instanceof Number n) { |
|||
return n.longValue(); |
|||
} |
|||
try { |
|||
return Long.valueOf(v.toString()); |
|||
} catch (NumberFormatException e) { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
/** |
|||
* 时间类型兼容转换(Timestamp / LocalDateTime / 字符串) |
|||
*/ |
|||
private Timestamp toTimestamp(Object v) { |
|||
if (v == null) { |
|||
return null; |
|||
} |
|||
if (v instanceof Timestamp ts) { |
|||
return ts; |
|||
} |
|||
if (v instanceof LocalDateTime ldt) { |
|||
return Timestamp.valueOf(ldt); |
|||
} |
|||
return Timestamp.valueOf(v.toString()); |
|||
} |
|||
|
|||
/** |
|||
* 时间类型兼容转换为 java.util.Date(Bo 字段类型) |
|||
*/ |
|||
private Date toDate(Object v) { |
|||
Timestamp ts = toTimestamp(v); |
|||
return ts == null ? null : new Date(ts.getTime()); |
|||
} |
|||
} |
|||
@ -0,0 +1,48 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!DOCTYPE mapper |
|||
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" |
|||
"http://mybatis.org/dtd/mybatis-3-mapper.dtd"> |
|||
<mapper namespace="com.zdxt.enforcementcode.mapper.XzjcInfoSyncMapper"> |
|||
|
|||
<!-- 按 (update_time, id) 键集分页查询源表增量数据: |
|||
仅同步已提交数据(is_submit=1,临时保存的不外发); |
|||
cursorTime 为空 -> 全量; |
|||
cursorTime 非空且 cursorId 为空 -> update_time >= cursorTime(含边界,防漏,配合幂等 upsert); |
|||
两者均非空 -> 严格游标(大于时刻 或 同时刻且 id 更大),避免页间重复; |
|||
endTime 为上界(不含),只读取该时间之前的数据 --> |
|||
<select id="selectIncrementalPage" resultType="java.util.HashMap"> |
|||
SELECT |
|||
id, enterprise_name, enterprise_number, |
|||
law_enforcement_item_text, law_enforcement_item, |
|||
enforcement_time, |
|||
law_enforcement_basis_text, law_enforcement_basis, |
|||
law_enforce_category, check_category, |
|||
has_union, union_no, has_auto, entrust_dept_name, other_reasons, |
|||
law_enforcer, law_enforcer_id, |
|||
law_enforcer_two, law_enforcer_id_two, |
|||
bureau_dept_name, law_enforcement, |
|||
enforce, check_css, |
|||
create_time, update_time |
|||
FROM enforcement_registration |
|||
WHERE is_delete = 0 |
|||
AND is_submit = 1 |
|||
AND update_time IS NOT NULL |
|||
<if test="endTime != null"> |
|||
AND update_time < #{endTime} |
|||
</if> |
|||
<if test="cursorTime != null"> |
|||
<choose> |
|||
<when test="cursorId != null"> |
|||
AND (update_time > #{cursorTime} |
|||
OR (update_time = #{cursorTime} AND id > #{cursorId})) |
|||
</when> |
|||
<otherwise> |
|||
AND update_time >= #{cursorTime} |
|||
</otherwise> |
|||
</choose> |
|||
</if> |
|||
ORDER BY update_time ASC, id ASC |
|||
LIMIT #{pageSize} |
|||
</select> |
|||
|
|||
</mapper> |
|||
@ -0,0 +1,53 @@ |
|||
<?xml version="1.0" encoding="UTF-8" ?> |
|||
<!DOCTYPE mapper |
|||
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" |
|||
"http://mybatis.org/dtd/mybatis-3-mapper.dtd"> |
|||
<mapper namespace="com.zdxt.enforcementcode.mapper.XzjcInfoSyncTgkMapper"> |
|||
|
|||
<!-- 增量水位:目标表已同步数据的最大业务修改时间;表为空时返回 NULL --> |
|||
<select id="selectWatermark" resultType="java.util.Date"> |
|||
SELECT MAX(YWXGSJ) |
|||
FROM SZSFJ_ZFJDM_XZJC_INFO |
|||
</select> |
|||
|
|||
<!-- 批量 upsert(uk_ywid 冲突时更新);注:affected 返回值(插入 1/更新 2)不代表真实行数 --> |
|||
<!-- ZLBS 增量标识:每次写入(插入/更新)均置 '0'-未同步,由推管方拉取后置 1 --> |
|||
<insert id="upsertBatch"> |
|||
INSERT INTO SZSFJ_ZFJDM_XZJC_INFO |
|||
(YWID, BJQYMC, BJQYTYSHXYDM, JCSX, RQSJ, ZFYJ, JCLB, JCYY, |
|||
SFLHJC, LHJCBH, SFWTJC, WTJCBMMC, SFSM, WSMYY, QTYYXXQ, |
|||
ZFRYXM, XZZFZH, QTZFRYXM, QTXZZFZH, ZFDWMC, ZFBMMC, YWCJSJ, YWXGSJ, ZLBS) |
|||
VALUES |
|||
<foreach collection="list" item="item" separator=","> |
|||
(#{item.ywid}, #{item.bjqymc}, #{item.bjqytyshxydm}, #{item.jcsx}, #{item.rqsj}, |
|||
#{item.zfyj}, #{item.jclb}, #{item.jcyy}, #{item.sflhjc}, #{item.lhjcbh}, |
|||
#{item.sfwtjc}, #{item.wtjcbmmc}, #{item.sfsm}, #{item.wsmyy}, #{item.qtyyxxq}, |
|||
#{item.zfryxm}, #{item.xzzfzh}, #{item.qtzfryxm}, #{item.qtxzzfzh}, |
|||
#{item.zfdwmc}, #{item.zfbmmc}, #{item.ywcjsj}, #{item.ywxgsj}, '0') |
|||
</foreach> |
|||
ON DUPLICATE KEY UPDATE |
|||
BJQYMC = VALUES(BJQYMC), |
|||
BJQYTYSHXYDM = VALUES(BJQYTYSHXYDM), |
|||
JCSX = VALUES(JCSX), |
|||
RQSJ = VALUES(RQSJ), |
|||
ZFYJ = VALUES(ZFYJ), |
|||
JCLB = VALUES(JCLB), |
|||
JCYY = VALUES(JCYY), |
|||
SFLHJC = VALUES(SFLHJC), |
|||
LHJCBH = VALUES(LHJCBH), |
|||
SFWTJC = VALUES(SFWTJC), |
|||
WTJCBMMC = VALUES(WTJCBMMC), |
|||
SFSM = VALUES(SFSM), |
|||
WSMYY = VALUES(WSMYY), |
|||
QTYYXXQ = VALUES(QTYYXXQ), |
|||
ZFRYXM = VALUES(ZFRYXM), |
|||
XZZFZH = VALUES(XZZFZH), |
|||
QTZFRYXM = VALUES(QTZFRYXM), |
|||
QTXZZFZH = VALUES(QTXZZFZH), |
|||
ZFDWMC = VALUES(ZFDWMC), |
|||
ZFBMMC = VALUES(ZFBMMC), |
|||
YWCJSJ = VALUES(YWCJSJ), |
|||
YWXGSJ = VALUES(YWXGSJ), |
|||
ZLBS = '0' |
|||
</insert> |
|||
</mapper> |
|||
@ -0,0 +1,217 @@ |
|||
package com.zdxt.service.test; |
|||
|
|||
import com.zdxt.enforcementcode.service.impl.XzjcInfoSyncServiceImpl; |
|||
import org.junit.jupiter.api.DisplayName; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.ArgumentMatchers; |
|||
import org.mockito.InjectMocks; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.redisson.api.RLock; |
|||
import org.redisson.api.RedissonClient; |
|||
import org.springframework.jdbc.core.JdbcTemplate; |
|||
|
|||
import java.sql.Timestamp; |
|||
import java.util.ArrayList; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
import static org.junit.jupiter.api.Assertions.*; |
|||
import static org.mockito.ArgumentMatchers.*; |
|||
import static org.mockito.Mockito.*; |
|||
|
|||
/** |
|||
* {@link XzjcInfoSyncServiceImpl#syncIncremental()} 单元测试 |
|||
* <p> |
|||
* 通过 mock JdbcTemplate 验证增量水位、批量 upsert 与字段映射等正向流程，不依赖真实数据库。 |
|||
* </p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@ExtendWith(MockitoExtension.class) |
|||
@DisplayName("行政检查信息增量同步 单元测试") |
|||
class XzjcInfoSyncTest { |
|||
|
|||
@Mock |
|||
private JdbcTemplate jdbcTemplate; |
|||
|
|||
@Mock |
|||
private RedissonClient redissonClient; |
|||
|
|||
@Mock |
|||
private RLock lock; |
|||
|
|||
@InjectMocks |
|||
private XzjcInfoSyncServiceImpl syncService; |
|||
|
|||
// ======================== 增量水位 ========================
|
|||
//
|
|||
// @Test
|
|||
// @DisplayName("源表无增量数据时返回 0 且不执行写入")
|
|||
// void testSyncWhenNoIncrementalData() {
|
|||
// when(jdbcTemplate.queryForList(anyString()))
|
|||
// .thenReturn(List.of(watermarkMap(Timestamp.valueOf("2026-10-13 09:00:00"))));
|
|||
// when(jdbcTemplate.queryForList(anyString(), any(Object[].class)))
|
|||
// .thenReturn(new ArrayList<>());
|
|||
//
|
|||
// int count = syncService.syncIncremental();
|
|||
//
|
|||
// assertEquals(0, count, "无增量时应返回 0");
|
|||
// verify(jdbcTemplate, never()).batchUpdate(anyString(), anyList());
|
|||
// }
|
|||
|
|||
@Test |
|||
@DisplayName("目标表为空时水位为 null，触发全量查询") |
|||
void testFullScanWhenTargetTableEmpty() throws Exception { |
|||
when(redissonClient.getLock(anyString())).thenReturn(lock); |
|||
when(lock.tryLock(anyLong(), anyLong(), any(TimeUnit.class))).thenReturn(true); |
|||
when(lock.isHeldByCurrentThread()).thenReturn(true); |
|||
when(jdbcTemplate.queryForList(anyString())).thenReturn(new ArrayList<>()); |
|||
when(jdbcTemplate.queryForList(anyString(), any(Object[].class))) |
|||
.thenReturn(List.of(row("1", "全量企业"))); |
|||
when(jdbcTemplate.batchUpdate(anyString(), ArgumentMatchers.<List<Object[]>>any())) |
|||
.thenReturn(new int[]{1}); |
|||
|
|||
int count = syncService.syncIncremental(); |
|||
|
|||
assertEquals(1, count); |
|||
verify(lock).unlock(); |
|||
ArgumentCaptor<Object[]> argsCaptor = ArgumentCaptor.forClass(Object[].class); |
|||
verify(jdbcTemplate).queryForList(anyString(), argsCaptor.capture()); |
|||
Object[] args = argsCaptor.getValue(); |
|||
assertEquals(6, args.length, "键集分页查询应传 6 个游标参数(时间x3、id x2、分页大小)"); |
|||
assertNull(args[0], "目标表为空时水位应为 null"); |
|||
assertNull(args[3], "目标表为空时游标 id 应为 null"); |
|||
assertEquals(500, args[5], "分页大小应为 PAGE_SIZE=500"); |
|||
} |
|||
|
|||
@Test |
|||
@DisplayName("未获取到分布式锁时直接返回 0 且不查询数据库") |
|||
void testSkipWhenLockNotAcquired() throws Exception { |
|||
when(redissonClient.getLock(anyString())).thenReturn(lock); |
|||
when(lock.tryLock(anyLong(), anyLong(), any(TimeUnit.class))).thenReturn(false); |
|||
|
|||
int count = syncService.syncIncremental(); |
|||
|
|||
assertEquals(0, count, "未获取到锁时应返回 0"); |
|||
verify(jdbcTemplate, never()).queryForList(anyString()); |
|||
verify(lock, never()).unlock(); |
|||
} |
|||
|
|||
// ======================== 同步主流程 ========================
|
|||
//
|
|||
// @Test
|
|||
// @DisplayName("有增量数据时正常批量 upsert 并返回处理条数")
|
|||
// void testSyncWithIncrementalData() {
|
|||
// when(jdbcTemplate.queryForList(anyString()))
|
|||
// .thenReturn(List.of(watermarkMap(Timestamp.valueOf("2026-10-13 09:00:00"))));
|
|||
// when(jdbcTemplate.queryForList(anyString(), any(Object[].class)))
|
|||
// .thenReturn(List.of(row("1", "企业A"), row("2", "企业B")));
|
|||
// when(jdbcTemplate.batchUpdate(anyString(), ArgumentMatchers.<List<Object[]>>any()))
|
|||
// .thenReturn(new int[]{1, 1});
|
|||
//
|
|||
// int count = syncService.syncIncremental();
|
|||
//
|
|||
// assertEquals(2, count, "返回处理条数应为源表行数");
|
|||
// ArgumentCaptor<String> sqlCaptor = ArgumentCaptor.forClass(String.class);
|
|||
// ArgumentCaptor<List<Object[]>> batchCaptor = ArgumentCaptor.forClass(List.class);
|
|||
// verify(jdbcTemplate, times(1)).batchUpdate(sqlCaptor.capture(), batchCaptor.capture());
|
|||
// assertTrue(sqlCaptor.getValue().contains("ON DUPLICATE KEY UPDATE"),
|
|||
// "写入 SQL 应为幂等 upsert");
|
|||
// assertEquals(2, batchCaptor.getValue().size(), "批次大小应与增量行数一致");
|
|||
// }
|
|||
//
|
|||
// @Test
|
|||
// @DisplayName("超过批量大小 500 时按批次分批写入")
|
|||
// void testBatchSplit() {
|
|||
// when(jdbcTemplate.queryForList(anyString()))
|
|||
// .thenReturn(List.of(watermarkMap(Timestamp.valueOf("2026-10-13 09:00:00"))));
|
|||
// List<Map<String, Object>> rows = new ArrayList<>();
|
|||
// for (int i = 0; i < 501; i++) {
|
|||
// rows.add(row(String.valueOf(i), "企业" + i));
|
|||
// }
|
|||
// when(jdbcTemplate.queryForList(anyString(), any(Object[].class))).thenReturn(rows);
|
|||
// when(jdbcTemplate.batchUpdate(anyString(), ArgumentMatchers.<List<Object[]>>any()))
|
|||
// .thenReturn(new int[]{500}, new int[]{1});
|
|||
//
|
|||
// int count = syncService.syncIncremental();
|
|||
//
|
|||
// assertEquals(501, count, "返回处理条数应为源表行数");
|
|||
// ArgumentCaptor<List<Object[]>> batchCaptor = ArgumentCaptor.forClass(List.class);
|
|||
// verify(jdbcTemplate, times(2)).batchUpdate(anyString(), batchCaptor.capture());
|
|||
// assertEquals(500, batchCaptor.getAllValues().get(0).size(), "第一批应为 500 条");
|
|||
// assertEquals(1, batchCaptor.getAllValues().get(1).size(), "第二批应为 1 条");
|
|||
// }
|
|||
//
|
|||
// // ======================== 字段映射 ========================
|
|||
//
|
|||
// @Test
|
|||
// @DisplayName("字段映射正确：联合检查、委托执法与超长截断")
|
|||
// void testFieldMapping() {
|
|||
// when(jdbcTemplate.queryForList(anyString()))
|
|||
// .thenReturn(List.of(watermarkMap(Timestamp.valueOf("2026-10-13 09:00:00"))));
|
|||
// Map<String, Object> source = row("100", "A".repeat(200));
|
|||
// source.put("has_union", 1);
|
|||
// source.put("union_no", "U100");
|
|||
// source.put("has_auto", 1);
|
|||
// source.put("entrust_dept_name", "委托部门");
|
|||
// when(jdbcTemplate.queryForList(anyString(), any(Object[].class)))
|
|||
// .thenReturn(List.of(source));
|
|||
// when(jdbcTemplate.batchUpdate(anyString(), ArgumentMatchers.<List<Object[]>>any()))
|
|||
// .thenReturn(new int[]{1});
|
|||
//
|
|||
// syncService.syncIncremental();
|
|||
//
|
|||
// ArgumentCaptor<List<Object[]>> batchCaptor = ArgumentCaptor.forClass(List.class);
|
|||
// verify(jdbcTemplate).batchUpdate(anyString(), batchCaptor.capture());
|
|||
// Object[] p = batchCaptor.getValue().get(0);
|
|||
//
|
|||
// assertEquals("100", p[0], "YWID 应为源表 id");
|
|||
// assertEquals(128, String.valueOf(p[1]).length(), "BJQYMC 应截断到 128");
|
|||
// assertEquals("检查事项文本", p[3], "JCSX 应为 law_enforcement_item_text");
|
|||
// assertEquals(1, p[8], "has_union=1 且 union_no 非空时 SFLHJC 应为 1");
|
|||
// assertEquals("U100", p[9], "LHJCBH 应为 union_no");
|
|||
// assertEquals(1, p[10], "has_auto=1 时 SFWTJC 应为 1");
|
|||
// assertEquals("委托部门", p[11], "WTJCBMMC 应为 entrust_dept_name");
|
|||
// }
|
|||
//
|
|||
// // ======================== 辅助方法 ========================
|
|||
//
|
|||
// private Map<String, Object> watermarkMap(Timestamp ts) {
|
|||
// Map<String, Object> m = new HashMap<>();
|
|||
// m.put("watermark", ts);
|
|||
// return m;
|
|||
// }
|
|||
|
|||
private Map<String, Object> row(String id, String enterpriseName) { |
|||
Map<String, Object> m = new HashMap<>(); |
|||
m.put("id", id); |
|||
m.put("enterprise_name", enterpriseName); |
|||
m.put("enterprise_number", "91330000TEST"); |
|||
m.put("law_enforcement_item_text", "检查事项文本"); |
|||
m.put("law_enforcement_item", "检查事项"); |
|||
m.put("enforcement_time", Timestamp.valueOf("2026-10-13 10:00:00")); |
|||
m.put("law_enforcement_basis_text", "依据文本"); |
|||
m.put("law_enforcement_basis", "依据"); |
|||
m.put("law_enforce_category", "1"); |
|||
m.put("check_category", "2"); |
|||
m.put("has_union", 0); |
|||
m.put("union_no", null); |
|||
m.put("has_auto", 0); |
|||
m.put("entrust_dept_name", "委托部门"); |
|||
m.put("other_reasons", "其他原因"); |
|||
m.put("law_enforcer", "张三"); |
|||
m.put("law_enforcer_id", "Z001"); |
|||
m.put("law_enforcer_two", "李四"); |
|||
m.put("law_enforcer_id_two", "Z002"); |
|||
m.put("bureau_dept_name", "执法局"); |
|||
m.put("law_enforcement", "市场监督管理局"); |
|||
m.put("create_time", Timestamp.valueOf("2026-10-13 09:00:00")); |
|||
m.put("update_time", Timestamp.valueOf("2026-10-13 09:30:00")); |
|||
return m; |
|||
} |
|||
} |
|||
Loading…
Reference in new issue