list);
+}
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 59e5eaf1..33b87394 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
@@ -1,22 +1,32 @@
package com.zdxt.enforcementcode.service.impl;
+import com.zdxt.enforcementcode.domain.bo.XzjcInfoSyncBo;
+import com.zdxt.enforcementcode.mapper.XzjcInfoSyncMapper;
import com.zdxt.enforcementcode.service.IXzjcInfoSyncService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
-import org.springframework.jdbc.core.JdbcTemplate;
+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;
/**
* 行政检查信息同步服务
* 增量同步 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
*
* @author zdxt
*/
@@ -25,153 +35,144 @@ import java.util.Map;
@RequiredArgsConstructor
public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService {
- /** 批量写入大小 */
- private static final int BATCH_SIZE = 500;
+ /** 分页读取大小(与批量写入大小一致,一页一批) */
+ private static final int PAGE_SIZE = 500;
- private final JdbcTemplate jdbcTemplate;
+ /** 单次任务最大分页数,防止异常场景下扫描失控(500*1000=单次最多约 50 万条,剩余下次任务继续) */
+ private static final int MAX_PAGES = 1000;
+
+ /** 同步分布式锁 key */
+ private static final String LOCK_KEY = "xzjc_info_sync:lock";
+
+ private final XzjcInfoSyncMapper xzjcInfoSyncMapper;
+
+ private final RedissonClient redissonClient;
/**
* 执行一次增量同步
+ * 通过 Redis 分布式锁保证同一时刻仅一个任务实例在执行(含多实例部署场景)
*
- * @return 处理条数
+ * @return 处理条数;未获取到锁(上一次仍在执行)或被中断时返回 0
*/
public int syncIncremental() {
- Timestamp watermark = queryWatermark();
- List> rows = queryIncremental(watermark);
- if (rows.isEmpty()) {
- log.info("[XzjcInfoSync] 无增量数据, watermark={}", watermark);
+ 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;
- }
-
- int affected = 0;
- List batch = new ArrayList<>(BATCH_SIZE);
- for (Map row : rows) {
- batch.add(toParams(row));
- if (batch.size() >= BATCH_SIZE) {
- affected += flushBatch(batch);
- batch.clear();
+ } finally {
+ if (locked && lock.isHeldByCurrentThread()) {
+ lock.unlock();
}
}
- affected += flushBatch(batch);
-
- log.info("[XzjcInfoSync] 增量同步完成: 读取 {} 条, 写入 {} 条, watermark={}", rows.size(), affected, watermark);
- return rows.size();
}
/**
- * 增量水位:目标表中已同步数据的最大业务修改时间
+ * 分页增量同步主体
*/
- private Timestamp queryWatermark() {
- List> list = jdbcTemplate.queryForList(
- "SELECT MAX(YWXGSJ) AS watermark FROM SZSFJ_ZFJDM_XZJC_INFO");
- if (list.isEmpty() || list.get(0).get("watermark") == null) {
- return null;
+ private int doSync() {
+ // 数据截止时间:今天零点(即只同步前一天晚上 23:59:59 及更早的数据),避免同步当日仍在变化的数据
+ Date endTime = Timestamp.valueOf(LocalDate.now().atStartOfDay());
+ Date watermark = xzjcInfoSyncMapper.selectWatermark();
+ Date cursorTime = watermark;
+ String cursorId = null;
+ int total = 0;
+ boolean limitReached = false;
+
+ for (int page = 1; page <= MAX_PAGES; page++) {
+ List> rows = xzjcInfoSyncMapper.selectIncrementalPage(cursorTime, cursorId, endTime, PAGE_SIZE);
+ if (rows.isEmpty()) {
+ break;
+ }
+
+ List batch = new ArrayList<>(rows.size());
+ for (Map row : rows) {
+ batch.add(toBo(row));
+ }
+ upsertSafely(batch);
+ total += rows.size();
+
+ // 游标推进到本页最后一条,下页从其后继续
+ Map 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;
+ }
}
- Object v = list.get(0).get("watermark");
- return v instanceof Timestamp ts ? ts : Timestamp.valueOf(v.toString());
- }
- /**
- * 查询源表增量数据(水位为 NULL 时全量)
- */
- private List> queryIncremental(Timestamp watermark) {
- String sql = """
- 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,
- create_time, update_time
- FROM enforcement_registration
- WHERE is_delete = 0
- AND update_time IS NOT NULL
- AND (? IS NULL OR update_time >= ?)
- ORDER BY update_time ASC, id ASC
- """;
- return jdbcTemplate.queryForList(sql, watermark, watermark);
+ if (limitReached) {
+ log.warn("[XzjcInfoSync] 达到单次最大分页数 {}(约 {} 条), 剩余数据下次任务继续同步", MAX_PAGES, (long) MAX_PAGES * PAGE_SIZE);
+ }
+ log.info("[XzjcInfoSync] 增量同步完成: 共处理 {} 条, 起始水位={}, 截止={}", total, watermark, endTime);
+ return total;
}
/**
- * 批量 upsert(uk_ywid 冲突时更新)
+ * 页级批量 upsert;整批失败时降级为逐条写入,跳过脏数据行,避免单行异常导致游标卡死
*/
- private int flushBatch(List batch) {
- if (batch.isEmpty()) {
- return 0;
- }
- String sql = """
- INSERT INTO SZSFJ_ZFJDM_XZJC_INFO
- (YWID, BJQYMC, BJQYTYSHXYDM, JCSX, RQSJ, ZFYJ, JCLB, JCYY,
- SFLHJC, LHJCBH, SFWTJC, WTJCBMMC, QTYYXXQ,
- ZFRYXM, XZZFZH, QTZFRYXM, QTXZZFZH, ZFDWMC, ZFBMMC, YWCJSJ, YWXGSJ)
- VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
- 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),
- 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)
- """;
- int[] results = jdbcTemplate.batchUpdate(sql, batch);
- int affected = 0;
- for (int r : results) {
- affected += r;
+ private void upsertSafely(List batch) {
+ try {
+ xzjcInfoSyncMapper.upsertBatch(batch);
+ } catch (Exception e) {
+ log.error("[XzjcInfoSync] 批量写入失败,降级逐条写入排查脏数据, 批次大小={}", batch.size(), e);
+ for (XzjcInfoSyncBo bo : batch) {
+ try {
+ xzjcInfoSyncMapper.upsertBatch(List.of(bo));
+ } catch (Exception rowEx) {
+ log.error("[XzjcInfoSync] 脏数据跳过: ywid={}", bo.getYwid(), rowEx);
+ }
+ }
}
- return affected;
}
/**
- * 源表行 -> 目标表参数(字段映射与长度截断)
+ * 源表行 -> 目标表写入行(字段映射与长度截断)
*/
- private Object[] toParams(Map row) {
+ private XzjcInfoSyncBo toBo(Map 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.isEmpty();
-
- return new Object[]{
- str(row.get("id")),
- fit(row.get("enterprise_name"), 128),
- fit(row.get("enterprise_number"), 32),
- fit(coalesce(row, "law_enforcement_item_text", "law_enforcement_item"), 512),
- row.get("enforcement_time"),
- fit(coalesce(row, "law_enforcement_basis_text", "law_enforcement_basis"), 512),
- fit(row.get("law_enforce_category"), 2),
- fit(row.get("check_category"), 2),
- isUnion ? 1 : 0,
- fit(unionNo, 50),
- hasAuto != null && hasAuto == 1 ? 1 : 0,
- fit(row.get("entrust_dept_name"), 128),
- fit(row.get("other_reasons"), 512),
- fit(row.get("law_enforcer"), 64),
- fit(row.get("law_enforcer_id"), 64),
- fit(row.get("law_enforcer_two"), 64),
- fit(row.get("law_enforcer_id_two"), 64),
- fit(row.get("bureau_dept_name"), 64),
- fit(row.get("law_enforcement"), 64),
- row.get("create_time"),
- row.get("update_time")
- };
+ 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.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 row, String first, String second) {
@@ -204,4 +205,28 @@ public class XzjcInfoSyncServiceImpl implements IXzjcInfoSyncService {
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());
+ }
}
diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/resources/mapper/enforcementcode/XzjcInfoSyncMapper.xml b/zdxt-modules/zdxt-enforcement-code/src/main/resources/mapper/enforcementcode/XzjcInfoSyncMapper.xml
new file mode 100644
index 00000000..bfc265e5
--- /dev/null
+++ b/zdxt-modules/zdxt-enforcement-code/src/main/resources/mapper/enforcementcode/XzjcInfoSyncMapper.xml
@@ -0,0 +1,89 @@
+
+
+
+
+
+
+
+
+
+
+
+
+ INSERT INTO SZSFJ_ZFJDM_XZJC_INFO
+ (YWID, BJQYMC, BJQYTYSHXYDM, JCSX, RQSJ, ZFYJ, JCLB, JCYY,
+ SFLHJC, LHJCBH, SFWTJC, WTJCBMMC, QTYYXXQ,
+ ZFRYXM, XZZFZH, QTZFRYXM, QTXZZFZH, ZFDWMC, ZFBMMC, YWCJSJ, YWXGSJ)
+ VALUES
+
+ (#{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.qtyyxxq},
+ #{item.zfryxm}, #{item.xzzfzh}, #{item.qtzfryxm}, #{item.qtxzzfzh},
+ #{item.zfdwmc}, #{item.zfbmmc}, #{item.ywcjsj}, #{item.ywxgsj})
+
+ 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),
+ 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)
+
+
diff --git a/zdxt-web-server/zdxt-admin-web-server/src/test/java/com/zdxt/service/test/XzjcInfoSyncTest.java b/zdxt-web-server/zdxt-admin-web-server/src/test/java/com/zdxt/service/test/XzjcInfoSyncTest.java
index e269ec1f..e54f6856 100644
--- a/zdxt-web-server/zdxt-admin-web-server/src/test/java/com/zdxt/service/test/XzjcInfoSyncTest.java
+++ b/zdxt-web-server/zdxt-admin-web-server/src/test/java/com/zdxt/service/test/XzjcInfoSyncTest.java
@@ -9,6 +9,8 @@ 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;
@@ -16,6 +18,7 @@ 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.*;
@@ -36,6 +39,12 @@ class XzjcInfoSyncTest {
@Mock
private JdbcTemplate jdbcTemplate;
+ @Mock
+ private RedissonClient redissonClient;
+
+ @Mock
+ private RLock lock;
+
@InjectMocks
private XzjcInfoSyncServiceImpl syncService;
@@ -57,7 +66,10 @@ class XzjcInfoSyncTest {
@Test
@DisplayName("目标表为空时水位为 null，触发全量查询")
- void testFullScanWhenTargetTableEmpty() {
+ 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", "全量企业")));
@@ -67,15 +79,30 @@ class XzjcInfoSyncTest {
int count = syncService.syncIncremental();
assertEquals(1, count);
+ verify(lock).unlock();
ArgumentCaptor argsCaptor = ArgumentCaptor.forClass(Object[].class);
verify(jdbcTemplate).queryForList(anyString(), argsCaptor.capture());
Object[] args = argsCaptor.getValue();
- assertEquals(2, args.length, "增量查询应传 2 个水位参数");
+ assertEquals(6, args.length, "键集分页查询应传 6 个游标参数(时间x3、id x2、分页大小)");
assertNull(args[0], "目标表为空时水位应为 null");
- assertNull(args[1], "目标表为空时水位应为 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 并返回处理条数")