Browse Source

优化的编目的相关功能

dev-V1.10.1
vincent 2 weeks ago
parent
commit
603cf92a66
  1. 77
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/domain/bo/XzjcInfoSyncBo.java
  2. 46
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/mapper/XzjcInfoSyncMapper.java
  3. 259
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/service/impl/XzjcInfoSyncServiceImpl.java
  4. 89
      zdxt-modules/zdxt-enforcement-code/src/main/resources/mapper/enforcementcode/XzjcInfoSyncMapper.xml
  5. 37
      zdxt-web-server/zdxt-admin-web-server/src/test/java/com/zdxt/service/test/XzjcInfoSyncTest.java

77
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/domain/bo/XzjcInfoSyncBo.java

@ -0,0 +1,77 @@
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;
/** 其他原因详情(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;
}

46
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/mapper/XzjcInfoSyncMapper.java

@ -0,0 +1,46 @@
package com.zdxt.enforcementcode.mapper;
import com.zdxt.enforcementcode.domain.bo.XzjcInfoSyncBo;
import org.apache.ibatis.annotations.Param;
import java.util.Date;
import java.util.List;
import java.util.Map;
/**
* 行政检查信息同步 Mapper 接口
* <p>增量同步 enforcement_registration -> SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)</p>
*
* @author zdxt
*/
public interface XzjcInfoSyncMapper {
/**
* 查询增量水位:目标表已同步数据的最大业务修改时间(MAX(YWXGSJ))
*
* @return 水位时间;目标表为空时返回 null
*/
Date selectWatermark();
/**
* 按 (update_time, id) 键集分页查询源表增量数据
*
* @param cursorTime 游标时间(null 时全量)
* @param cursorId 游标 ID(cursorTime 非空且本参数为 null 时,按 update_time &gt;= 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);
/**
* 批量 upsert 目标表(基于 uk_ywid 唯一键 INSERT ... ON DUPLICATE KEY UPDATE,幂等)
*
* @param list 写入行集合
* @return 影响行数(MySQL upsert 语义,不代表真实写入条数)
*/
int upsertBatch(@Param("list") List<XzjcInfoSyncBo> list);
}

259
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;
/**
* 行政检查信息同步服务
* <p>增量同步 enforcement_registration -> SZSFJ_ZFJDM_XZJC_INFO(市司法局数据共享表)</p>
* <p>增量水位:目标表 MAX(YWXGSJ),源表 update_time >= 水位即为增量;目标表为空时首次自动全量</p>
* <p>写入策略:基于 uk_ywid 唯一键 INSERT ... ON DUPLICATE KEY UPDATE(幂等)</p>
* <p>读取策略:(update_time, id) 键集分页,单页单批,避免一次性加载全量数据导致内存溢出</p>
* <p>并发控制:Redisson 分布式锁,覆盖 Quartz 调度重叠、手动触发及多实例部署场景</p>
* <p>SQL 定义见 mapper/enforcementcode/XzjcInfoSyncMapper.xml</p>
*
* @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;
/**
* 执行一次增量同步
* <p>通过 Redis 分布式锁保证同一时刻仅一个任务实例在执行(含多实例部署场景)</p>
*
* @return 处理条数
* @return 处理条数;未获取到锁(上一次仍在执行)或被中断时返回 0
*/
public int syncIncremental() {
Timestamp watermark = queryWatermark();
List<Map<String, Object>> 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;
}
int affected = 0;
List<Object[]> batch = new ArrayList<>(BATCH_SIZE);
for (Map<String, Object> row : rows) {
batch.add(toParams(row));
if (batch.size() >= BATCH_SIZE) {
affected += flushBatch(batch);
batch.clear();
return doSync();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("[XzjcInfoSync] 同步任务获取锁时被中断", e);
return 0;
} 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<Map<String, Object>> 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<Map<String, Object>> rows = xzjcInfoSyncMapper.selectIncrementalPage(cursorTime, cursorId, endTime, PAGE_SIZE);
if (rows.isEmpty()) {
break;
}
Object v = list.get(0).get("watermark");
return v instanceof Timestamp ts ? ts : Timestamp.valueOf(v.toString());
List<XzjcInfoSyncBo> batch = new ArrayList<>(rows.size());
for (Map<String, Object> row : rows) {
batch.add(toBo(row));
}
upsertSafely(batch);
total += rows.size();
/**
* 查询源表增量数据(水位为 NULL 时全量)
*/
private List<Map<String, Object>> 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);
// 游标推进到本页最后一条,下页从其后继续
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] 增量同步完成: 共处理 {} 条, 起始水位={}, 截止={}", total, watermark, endTime);
return total;
}
/**
* 批量 upsert(uk_ywid 冲突时更新)
* 页级批量 upsert;整批失败时降级为逐条写入,跳过脏数据行,避免单行异常导致游标卡死
*/
private int flushBatch(List<Object[]> batch) {
if (batch.isEmpty()) {
return 0;
private void upsertSafely(List<XzjcInfoSyncBo> 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);
}
}
}
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;
}
return affected;
}
/**
* 源表行 -> 目标表参数(字段映射与长度截断)
* 源表行 -> 目标表写入行(字段映射与长度截断)
*/
private Object[] toParams(Map<String, Object> row) {
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.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<String, Object> 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());
}
}

89
zdxt-modules/zdxt-enforcement-code/src/main/resources/mapper/enforcementcode/XzjcInfoSyncMapper.xml

@ -0,0 +1,89 @@
<?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">
<!-- 增量水位:目标表已同步数据的最大业务修改时间;表为空时返回 NULL -->
<select id="selectWatermark" resultType="java.util.Date">
SELECT MAX(YWXGSJ)
FROM SZSFJ_ZFJDM_XZJC_INFO
</select>
<!-- 按 (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,
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 &lt; #{endTime}
</if>
<if test="cursorTime != null">
<choose>
<when test="cursorId != null">
AND (update_time &gt; #{cursorTime}
OR (update_time = #{cursorTime} AND id &gt; #{cursorId}))
</when>
<otherwise>
AND update_time &gt;= #{cursorTime}
</otherwise>
</choose>
</if>
ORDER BY update_time ASC, id ASC
LIMIT #{pageSize}
</select>
<!-- 批量 upsert(uk_ywid 冲突时更新);注:affected 返回值(插入 1/更新 2)不代表真实行数 -->
<insert id="upsertBatch">
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
<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.qtyyxxq},
#{item.zfryxm}, #{item.xzzfzh}, #{item.qtzfryxm}, #{item.qtxzzfzh},
#{item.zfdwmc}, #{item.zfbmmc}, #{item.ywcjsj}, #{item.ywxgsj})
</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),
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)
</insert>
</mapper>

37
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<Object[]> 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 并返回处理条数")

Loading…
Cancel
Save