vincent
1 hour ago
20 changed files with 671 additions and 113 deletions
@ -0,0 +1,42 @@ |
|||
CREATE TABLE SZ_SFJ_ZFJDM_XZJC_INFO ( |
|||
ID BIGINT NOT NULL IDENTITY(1,1) COMMENT '主键，自增ID', |
|||
YWID VARCHAR(64) NOT NULL COMMENT '业务ID', |
|||
BJQYMC VARCHAR(128) DEFAULT NULL COMMENT '被检查企业名称', |
|||
BJQYTYSHXYDM VARCHAR(32) DEFAULT NULL COMMENT '被检查企业统一社会信用代码', |
|||
SSQYBM VARCHAR(64) DEFAULT NULL COMMENT '所属区域编码', |
|||
SSQYMC VARCHAR(64) 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 '其他行政执法证号', |
|||
ZFDWID VARCHAR(64) DEFAULT NULL COMMENT '执法单位ID', |
|||
ZFDWMC VARCHAR(64) DEFAULT NULL COMMENT '执法单位名称', |
|||
ZFBMMC VARCHAR(64) DEFAULT NULL COMMENT '执法部门名称', |
|||
SFSC TINYINT DEFAULT 0 COMMENT '是否删除：0否，1是', |
|||
YWCJSJ DATETIME DEFAULT NULL COMMENT '业务创建时间', |
|||
YWXGSJ DATETIME DEFAULT NULL COMMENT '业务修改时间', |
|||
ADD_TIME DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', |
|||
CD_TIME DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '修改时间', |
|||
CD_OPERATION VARCHAR(4) DEFAULT NULL COMMENT '增量标识：I|插入，U|更新，D|删除', |
|||
CD_BATCH VARCHAR(32) DEFAULT NULL COMMENT '批次号', |
|||
REMARK VARCHAR(256) DEFAULT NULL COMMENT '备注', |
|||
PRIMARY KEY (ID) |
|||
) COMMENT='深圳市涉企行政执法扫码入企登记信息'; |
|||
|
|||
-- 索引 |
|||
CREATE INDEX IDX_BJQY_CREDIT ON SZ_SFJ_ZFJDM_XZJC_INFO(BJQYTYSHXYDM); |
|||
CREATE INDEX IDX_RQSJ ON SZ_SFJ_ZFJDM_XZJC_INFO(RQSJ); |
|||
-- YWID 唯一:同一业务ID只保留一行(先查后写的防重兜底) |
|||
CREATE UNIQUE INDEX UK_YWID ON SZ_SFJ_ZFJDM_XZJC_INFO(YWID); |
|||
@ -0,0 +1,18 @@ |
|||
package com.zdxt.enforcementcode.domain.bo.datashare; |
|||
|
|||
import com.zdxt.enforcementcode.domain.bo.XzjcInfoSyncBo; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
|
|||
/** |
|||
* 行政检查信息共享同步行(tgk 通知行,含行主键 ID 用于同步成功后回写 TBZT) |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@Data |
|||
@EqualsAndHashCode(callSuper = true) |
|||
public class XzjcInfoShareBo extends XzjcInfoSyncBo { |
|||
|
|||
/** tgk 表行主键(同步成功后按此回写 TBZT='1') */ |
|||
private Long id; |
|||
} |
|||
@ -0,0 +1,41 @@ |
|||
package com.zdxt.enforcementcode.mapper.datashare; |
|||
|
|||
import com.baomidou.dynamic.datasource.annotation.DS; |
|||
import com.zdxt.enforcementcode.domain.bo.datashare.XzjcInfoShareBo; |
|||
import org.apache.ibatis.annotations.Param; |
|||
|
|||
import java.util.List; |
|||
|
|||
/** |
|||
* 行政检查信息共享同步 Mapper 接口(zfjd_gxk 达梦目标侧) |
|||
* <p>YWID 唯一口径:不存在则插入(CD_OPERATION='I'),存在则更新(CD_OPERATION='U')</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@DS("gxk") |
|||
public interface XzjcInfoShareGxkMapper { |
|||
|
|||
/** |
|||
* 查询已存在于目标表的业务ID集合(用于 I/U 判定) |
|||
* |
|||
* @param list 待判定的业务ID集合 |
|||
* @return 目标表中已存在的业务ID |
|||
*/ |
|||
List<String> selectExistingYwids(@Param("list") List<String> list); |
|||
|
|||
/** |
|||
* 插入新行(目标表不存在该 YWID),CD_OPERATION 由调用方置 'I' |
|||
* |
|||
* @param item 写入行 |
|||
* @return 影响行数 |
|||
*/ |
|||
int insertOne(@Param("item") XzjcInfoShareBo item); |
|||
|
|||
/** |
|||
* 按业务ID更新已存在行(目标表存在该 YWID),CD_OPERATION 由调用方置 'U' |
|||
* |
|||
* @param item 写入行 |
|||
* @return 影响行数 |
|||
*/ |
|||
int updateByYwid(@Param("item") XzjcInfoShareBo item); |
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
package com.zdxt.enforcementcode.mapper.datashare; |
|||
|
|||
import com.baomidou.dynamic.datasource.annotation.DS; |
|||
import com.zdxt.enforcementcode.domain.bo.datashare.XzjcInfoShareBo; |
|||
import org.apache.ibatis.annotations.Param; |
|||
|
|||
import java.util.List; |
|||
|
|||
/** |
|||
* 行政检查信息共享同步 Mapper 接口(zfjd_tgk 源侧) |
|||
* <p>读取 TBZT='0' 的未拉取通知行;写入成功后回写 TBZT='1'</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@DS("tgk") |
|||
public interface XzjcInfoShareMapper { |
|||
|
|||
/** |
|||
* 分页查询未同步(TBZT='0')的通知行,按批次号顺序保证下游按序消费 |
|||
* |
|||
* @param pageSize 分页大小 |
|||
* @return 未同步行 |
|||
*/ |
|||
List<XzjcInfoShareBo> selectUnsyncedPage(@Param("pageSize") int pageSize); |
|||
|
|||
/** |
|||
* 回写同步状态:TBZT 置 '1'-已同步(限定 TBZT='0' 保证幂等) |
|||
* |
|||
* @param ids 已成功写入 gxk 的行主键集合 |
|||
* @return 影响行数 |
|||
*/ |
|||
int markSynced(@Param("ids") List<Long> ids); |
|||
} |
|||
@ -0,0 +1,35 @@ |
|||
package com.zdxt.enforcementcode.quartz.job; |
|||
|
|||
import com.zdxt.enforcementcode.service.IXzjcInfoShareService; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Component; |
|||
|
|||
/** |
|||
* 共享库同步到提供库（局级） |
|||
* 行政检查信息共享同步定时任务 |
|||
* <p>将 zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO 中 TBZT=0 的数据同步到 zfjd_gxk.SZ_SFJ_ZFJDM_XZJC_INFO</p> |
|||
* <p>默认 cron: 0 30 2 * * ?(每天凌晨 2:30,晚于 2:00 的 tgk 写入任务)</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@Slf4j |
|||
@Component |
|||
@RequiredArgsConstructor |
|||
public class XzjcInfoShareJob { |
|||
|
|||
private final IXzjcInfoShareService xzjcInfoShareService; |
|||
|
|||
/** |
|||
* 任务执行入口(public,供 JobInvokeUtil 通过反射调用) |
|||
*/ |
|||
public void execute() { |
|||
log.info("[XzjcInfoShareJob] 定时任务触发:行政检查信息共享同步(tgk->gxk)"); |
|||
try { |
|||
xzjcInfoShareService.shareSync(); |
|||
} catch (Exception e) { |
|||
log.error("[XzjcInfoShareJob] 行政检查信息共享同步失败", e); |
|||
throw e; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,16 @@ |
|||
package com.zdxt.enforcementcode.service; |
|||
|
|||
/** |
|||
* 行政检查信息共享同步服务(zfjd_tgk -> zfjd_gxk) |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
public interface IXzjcInfoShareService { |
|||
|
|||
/** |
|||
* 执行一次共享同步:拉取 tgk 中 TBZT='0' 的通知行写入 gxk,成功后回写 TBZT='1' |
|||
* |
|||
* @return 实际同步条数 |
|||
*/ |
|||
int shareSync(); |
|||
} |
|||
@ -0,0 +1,115 @@ |
|||
package com.zdxt.enforcementcode.service.impl; |
|||
|
|||
import com.zdxt.enforcementcode.domain.bo.datashare.XzjcInfoShareBo; |
|||
import com.zdxt.enforcementcode.mapper.datashare.XzjcInfoShareGxkMapper; |
|||
import com.zdxt.enforcementcode.mapper.datashare.XzjcInfoShareMapper; |
|||
import com.zdxt.enforcementcode.service.IXzjcInfoShareService; |
|||
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.util.ArrayList; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Set; |
|||
import java.util.concurrent.TimeUnit; |
|||
|
|||
/** |
|||
* 行政检查信息共享同步服务实现 |
|||
* <p>源表:zfjd_tgk.SZSFJ_ZFJDM_XZJC_INFO(TBZT='0');目标表:zfjd_gxk.SZ_SFJ_ZFJDM_XZJC_INFO</p> |
|||
* <p>写入成功的行回写源表 TBZT='1';单行失败跳过不标记,下次任务重试</p> |
|||
* |
|||
* @author zdxt |
|||
*/ |
|||
@Slf4j |
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class XzjcInfoShareServiceImpl implements IXzjcInfoShareService { |
|||
|
|||
/** 分页读取大小(一页一批) */ |
|||
private static final int PAGE_SIZE = 500; |
|||
|
|||
/** 单次任务最大分页数 */ |
|||
private static final int MAX_PAGES = 1000; |
|||
|
|||
/** 共享同步分布式锁 key */ |
|||
private static final String LOCK_KEY = "xzjc_info_share:lock"; |
|||
|
|||
private final XzjcInfoShareMapper xzjcInfoShareMapper; |
|||
private final XzjcInfoShareGxkMapper xzjcInfoShareGxkMapper; |
|||
private final RedissonClient redissonClient; |
|||
|
|||
@Override |
|||
public int shareSync() { |
|||
RLock lock = redissonClient.getLock(LOCK_KEY); |
|||
boolean locked = false; |
|||
try { |
|||
locked = lock.tryLock(0, -1, TimeUnit.SECONDS); |
|||
if (!locked) { |
|||
log.warn("[XzjcInfoShare] 上一次共享同步尚未结束(锁被占用),本次触发跳过"); |
|||
return 0; |
|||
} |
|||
return doShare(); |
|||
} catch (InterruptedException e) { |
|||
Thread.currentThread().interrupt(); |
|||
log.error("[XzjcInfoShare] 共享同步任务获取锁时被中断", e); |
|||
return 0; |
|||
} finally { |
|||
if (locked && lock.isHeldByCurrentThread()) { |
|||
lock.unlock(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private int doShare() { |
|||
int total = 0; |
|||
for (int page = 1; page <= MAX_PAGES; page++) { |
|||
List<XzjcInfoShareBo> rows = xzjcInfoShareMapper.selectUnsyncedPage(PAGE_SIZE); |
|||
if (rows.isEmpty()) { |
|||
break; |
|||
} |
|||
// 达梦目标表已存在的业务ID -> U,不存在 -> I(YWID 唯一口径)
|
|||
Set<String> existing = new HashSet<>(); |
|||
if (!rows.isEmpty()) { |
|||
List<String> ywids = new ArrayList<>(rows.size()); |
|||
for (XzjcInfoShareBo bo : rows) { |
|||
ywids.add(bo.getYwid()); |
|||
} |
|||
List<String> hit = xzjcInfoShareGxkMapper.selectExistingYwids(ywids); |
|||
if (hit != null) { |
|||
existing.addAll(hit); |
|||
} |
|||
} |
|||
// 逐行写入,仅成功行参与回写 TBZT,失败行下轮重试
|
|||
List<Long> syncedIds = new ArrayList<>(rows.size()); |
|||
for (XzjcInfoShareBo bo : rows) { |
|||
boolean isUpdate = existing.contains(bo.getYwid()); |
|||
bo.setCdOperation(isUpdate ? "U" : "I"); |
|||
try { |
|||
if (isUpdate) { |
|||
xzjcInfoShareGxkMapper.updateByYwid(bo); |
|||
} else { |
|||
xzjcInfoShareGxkMapper.insertOne(bo); |
|||
} |
|||
syncedIds.add(bo.getId()); |
|||
} catch (Exception rowEx) { |
|||
log.error("[XzjcInfoShare] 脏数据跳过: id={}, ywid={}, cd_operation={}", |
|||
bo.getId(), bo.getYwid(), bo.getCdOperation(), rowEx); |
|||
} |
|||
} |
|||
if (!syncedIds.isEmpty()) { |
|||
xzjcInfoShareMapper.markSynced(syncedIds); |
|||
} |
|||
total += syncedIds.size(); |
|||
log.info("[XzjcInfoShare] 第 {} 页完成, 本页成功 {}/{} 条", page, syncedIds.size(), rows.size()); |
|||
|
|||
if (rows.size() < PAGE_SIZE) { |
|||
break; |
|||
} |
|||
} |
|||
log.info("[XzjcInfoShare] 共享同步完成(zfjd_tgk -> zfjd_gxk): 共处理 {} 条", total); |
|||
return total; |
|||
} |
|||
} |
|||
@ -0,0 +1,67 @@ |
|||
<?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.datashare.XzjcInfoShareGxkMapper"> |
|||
|
|||
<!-- 已存在业务ID查询:I/U 判定依据 --> |
|||
<select id="selectExistingYwids" resultType="java.lang.String"> |
|||
SELECT YWID |
|||
FROM SZ_SFJ_ZFJDM_XZJC_INFO |
|||
WHERE YWID IN |
|||
<foreach collection="list" item="id" open="(" separator="," close=")"> |
|||
#{id} |
|||
</foreach> |
|||
</select> |
|||
|
|||
<!-- 插入新行(不存在该 YWID -> I);ADD_TIME 由数据库默认值维护 --> |
|||
<insert id="insertOne"> |
|||
INSERT INTO SZ_SFJ_ZFJDM_XZJC_INFO |
|||
(YWID, BJQYMC, BJQYTYSHXYDM, SSQYBM, SSQYMC, JCSX, RQSJ, ZFYJ, JCLB, JCYY, |
|||
SFLHJC, LHJCBH, SFWTJC, WTJCBMMC, SFSM, WSMYY, QTYYXXQ, |
|||
ZFRYXM, XZZFZH, QTZFRYXM, QTXZZFZH, ZFDWID, ZFDWMC, ZFBMMC, SFSC, YWCJSJ, YWXGSJ, |
|||
CD_OPERATION, CD_BATCH) |
|||
VALUES |
|||
(#{item.ywid}, #{item.bjqymc}, #{item.bjqytyshxydm}, #{item.ssqybm}, #{item.ssqymc}, #{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.zfdwid}, #{item.zfdwmc}, #{item.zfbmmc}, #{item.sfsc}, #{item.ywcjsj}, #{item.ywxgsj}, |
|||
#{item.cdOperation}, #{item.cdBatch}) |
|||
</insert> |
|||
|
|||
<!-- 按业务ID更新已存在行(存在该 YWID -> U);CD_BATCH/CD_OPERATION 刷新为最新批次通知再次拉取 --> |
|||
<update id="updateByYwid"> |
|||
UPDATE SZ_SFJ_ZFJDM_XZJC_INFO |
|||
SET BJQYMC = #{item.bjqymc}, |
|||
BJQYTYSHXYDM = #{item.bjqytyshxydm}, |
|||
SSQYBM = #{item.ssqybm}, |
|||
SSQYMC = #{item.ssqymc}, |
|||
JCSX = #{item.jcsx}, |
|||
RQSJ = #{item.rqsj}, |
|||
ZFYJ = #{item.zfyj}, |
|||
JCLB = #{item.jclb}, |
|||
JCYY = #{item.jcyy}, |
|||
SFLHJC = #{item.sflhjc}, |
|||
LHJCBH = #{item.lhjcbh}, |
|||
SFWTJC = #{item.sfwtjc}, |
|||
WTJCBMMC = #{item.wtjcbmmc}, |
|||
SFSM = #{item.sfsm}, |
|||
WSMYY = #{item.wsmyy}, |
|||
QTYYXXQ = #{item.qtyyxxq}, |
|||
ZFRYXM = #{item.zfryxm}, |
|||
XZZFZH = #{item.xzzfzh}, |
|||
QTZFRYXM = #{item.qtzfryxm}, |
|||
QTXZZFZH = #{item.qtxzzfzh}, |
|||
ZFDWID = #{item.zfdwid}, |
|||
ZFDWMC = #{item.zfdwmc}, |
|||
ZFBMMC = #{item.zfbmmc}, |
|||
SFSC = #{item.sfsc}, |
|||
YWCJSJ = #{item.ywcjsj}, |
|||
YWXGSJ = #{item.ywxgsj}, |
|||
CD_OPERATION = #{item.cdOperation}, |
|||
CD_BATCH = #{item.cdBatch}, |
|||
CD_TIME = CURRENT_TIMESTAMP |
|||
WHERE YWID = #{item.ywid} |
|||
</update> |
|||
</mapper> |
|||
@ -0,0 +1,29 @@ |
|||
<?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.datashare.XzjcInfoShareMapper"> |
|||
|
|||
<!-- 未同步通知行:TBZT='0',按批次号+行号顺序,保证下游按序消费 --> |
|||
<select id="selectUnsyncedPage" resultType="com.zdxt.enforcementcode.domain.bo.datashare.XzjcInfoShareBo"> |
|||
SELECT ID, YWID, BJQYMC, BJQYTYSHXYDM, SSQYBM, SSQYMC, JCSX, RQSJ, ZFYJ, JCLB, JCYY, |
|||
SFLHJC, LHJCBH, SFWTJC, WTJCBMMC, SFSM, WSMYY, QTYYXXQ, |
|||
ZFRYXM, XZZFZH, QTZFRYXM, QTXZZFZH, ZFDWID, ZFDWMC, ZFBMMC, SFSC, YWCJSJ, YWXGSJ, |
|||
CD_OPERATION, CD_BATCH |
|||
FROM SZSFJ_ZFJDM_XZJC_INFO |
|||
WHERE TBZT = '0' |
|||
ORDER BY CD_BATCH ASC, ID ASC |
|||
LIMIT #{pageSize} |
|||
</select> |
|||
|
|||
<!-- 回写已同步:限定 TBZT='0' 防并发重复标记 --> |
|||
<update id="markSynced"> |
|||
UPDATE SZSFJ_ZFJDM_XZJC_INFO |
|||
SET TBZT = '1' |
|||
WHERE TBZT = '0' |
|||
AND ID IN |
|||
<foreach collection="ids" item="id" open="(" separator="," close=")"> |
|||
#{id} |
|||
</foreach> |
|||
</update> |
|||
</mapper> |
|||
Loading…
Reference in new issue