Skip to content

H
huazheng-project-flink
提交 cc2dc68e 作者: think
操作

表vbak,vbap的一次性任务

上级 153d6b53
正在显示 包含 143 行增加 和 0 行删除
src/main/java/com/huazheng/project/greenplum/mapper/GPMapper.java
......@@ -327,6 +327,7 @@ public interface GPMapper {
@CacheEvict(key = "'selectVbak'+':'+#p0.vbeln+','+#p0.mandt")
public void deleteVbak(Vbak item);
public List<Vbak> selectVbakCheck(Vbak build);
public Long selectVbakMaxRowNum();
@Cacheable(key = "#root.method.name+':'+#p0.vbeln+','+#p0.posnr+','+#p0.etenr+','+#p0.mandt", unless="#result == null")
public Vbep selectVbep(Vbep vbep); // 查询替代删除
......@@ -442,6 +443,7 @@ public interface GPMapper {
public void deleteVbap(Vbap item);
public List<Vbap> selectVbapCheck(Vbap build);
public List<Vbap> selectVbapCheckByUpdate(Vbap build);
public Long selectVbapMaxRowNum();
......
src/main/java/com/huazheng/project/greenplum/service/impl/DeleteUpdateJobServiceImpl.java
......@@ -28,6 +28,8 @@ import com.huazheng.project.hana.model.T001w;
import com.huazheng.project.hana.model.T023t;
import com.huazheng.project.hana.model.Tspat;
import com.huazheng.project.hana.model.Tvkbt;
import com.huazheng.project.hana.model.Vbak;
import com.huazheng.project.hana.model.Vbap;
import com.huazheng.project.mssql.mapper.CrmMapper;
import com.huazheng.project.mssql.model.SampleApplicationProcess;
import com.huazheng.project.mssql2.mapper.HzcrmMapper;
......@@ -1162,4 +1164,116 @@ public class DeleteUpdateJobServiceImpl {
}
}
public void selectVbakCheck() {
try {
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:check:Vbak:rowNum", "0");
String rowNum = opsForValue.get("huazheng:check:Vbak:rowNum");
// 一次性任务的终止位
if (redis1Template.hasKey("huazheng:check:Vbak:stopNum") == false) {
// 查询最大计数器
String maxRowNum = gpMapper.selectVbakMaxRowNum().toString();
opsForValue.setIfAbsent("huazheng:check:Vbak:stopNum", maxRowNum);
}
Long stopNum = Long.parseLong(opsForValue.get("huazheng:check:Vbak:stopNum"));
if (stopNum < Long.parseLong(rowNum)) {
JobKey jobKey = JobKey.jobKey("selectVbakCheck_once", "DEFAULT"); // 删除任务
quartzScheduler.deleteJob(jobKey);
return;
}
Vbak build = Vbak.builder().rowNum(rowNum).build();
List<Vbak> list = gpMapper.selectVbakCheck(build); // 从数仓中查询一组数据
if (list.size() == 0) {
JobKey jobKey = JobKey.jobKey("selectVbakCheck_once", "DEFAULT"); // 删除任务
quartzScheduler.deleteJob(jobKey);
return;
}
list.forEach(target -> { // 遍历要检查的数据
Vbak source = sapMapper.selectVbakById(target); // 根据主键查询源库中的数据
String operator = "none";
if (source == null) { // 如果源库中没有数据
gpMapper.deleteVbak(target); // 删除数仓中的数据
operator = "delete";
} else { // 源库中有数据
String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
String thash = target.getHashResult(); // 数仓中数据的hash结果
if (!shash.equals(thash)) { // 如果hash结果不一致
source.setHashResult(shash);
while (true) {
try {
gpMapper.updateVbak(source); // 更新数据到数仓中
break;
} catch (RuntimeException e) {
log.error(e.getMessage());ThreadUtil.safeSleep(500);
}
}
ThreadUtil.safeSleep(500);
}
}
redis1Template.opsForValue().set("huazheng:check:Vbak:rowNum", target.getRowNum());
if (!operator.equals("none")) {
log.info(String.format("selectVbakCheck --> rowNum:%s, operator:%s", target.getRowNum(), operator));
}
});
} catch (Exception e) {
redis1Template.opsForValue().set("huazheng:checkError:Vbak:rowNum", getErrorInfoFromException(e));
}
}
public void selectVbapCheck() {
try {
ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
opsForValue.setIfAbsent("huazheng:check:Vbap:rowNum", "0");
String rowNum = opsForValue.get("huazheng:check:Vbap:rowNum");
// 一次性任务的终止位
if (redis1Template.hasKey("huazheng:check:Vbap:stopNum") == false) {
// 查询最大计数器
String maxRowNum = gpMapper.selectVbapMaxRowNum().toString();
opsForValue.setIfAbsent("huazheng:check:Vbap:stopNum", maxRowNum);
}
Long stopNum = Long.parseLong(opsForValue.get("huazheng:check:Vbap:stopNum"));
if (stopNum < Long.parseLong(rowNum)) {
JobKey jobKey = JobKey.jobKey("selectVbapCheck_once", "DEFAULT"); // 删除任务
quartzScheduler.deleteJob(jobKey);
return;
}
Vbap build = Vbap.builder().rowNum(rowNum).build();
List<Vbap> list = gpMapper.selectVbapCheck(build); // 从数仓中查询一组数据
if (list.size() == 0) {
JobKey jobKey = JobKey.jobKey("selectVbapCheck_once", "DEFAULT"); // 删除任务
quartzScheduler.deleteJob(jobKey);
return;
}
list.forEach(target -> { // 遍历要检查的数据
Vbap source = sapMapper.selectVbapById(target); // 根据主键查询源库中的数据
String operator = "none";
if (source == null) { // 如果源库中没有数据
gpMapper.deleteVbap(target); // 删除数仓中的数据
operator = "delete";
} else { // 源库中有数据
String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
String thash = target.getHashResult(); // 数仓中数据的hash结果
if (!shash.equals(thash)) { // 如果hash结果不一致
source.setHashResult(shash);
while (true) {
try {
gpMapper.updateVbap(source); // 更新数据到数仓中
break;
} catch (RuntimeException e) {
log.error(e.getMessage());ThreadUtil.safeSleep(500);
}
}
ThreadUtil.safeSleep(500);
}
}
redis1Template.opsForValue().set("huazheng:check:Vbap:rowNum", target.getRowNum());
if (!operator.equals("none")) {
log.info(String.format("selectVbapCheck --> rowNum:%s, operator:%s", target.getRowNum(), operator));
}
});
} catch (Exception e) {
redis1Template.opsForValue().set("huazheng:checkError:Vbap:rowNum", getErrorInfoFromException(e));
}
}
}
src/main/resources/mapper/greenplum/GPMapper_greenplum.xml
......@@ -1084,6 +1084,10 @@
<select id="selectVbakCheck" parameterType="com.huazheng.project.hana.model.Vbak" resultType="com.huazheng.project.hana.model.Vbak">
select * from Vbak where rownum &gt; #{rowNum} order by rownum limit 20
</select>
<select id="selectVbakMaxRowNum" resultType="long">
select max(rowNum) from Vbak
</select>
<select id="selectVbep" parameterType="com.huazheng.project.hana.model.Vbep" resultType="com.huazheng.project.hana.model.Vbep">
select * from vbep
......@@ -1434,6 +1438,9 @@
<select id="selectVbapCheckByUpdate" parameterType="com.huazheng.project.hana.model.Vbap" resultType="com.huazheng.project.hana.model.Vbap">
select * from Vbap where aedat &gt;= #{aedat} order by rownum
</select>
<select id="selectVbapMaxRowNum" resultType="long">
select max(rowNum) from Vbap
</select>
......
src/main/resources/spring-init.xml
......@@ -1271,6 +1271,26 @@
</property>
<property name="cronExpression" value="* * * * * ?" />
</bean>
<bean class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
<property name="jobDetail">
<bean parent="methodJobDetail">
<property name="name" value="selectVbakCheck_once"/>
<property name="targetObject" ref="deleteUpdateJobServiceImpl" />
<property name="targetMethod" value="selectVbakCheck" />
</bean>
</property>
<property name="cronExpression" value="* * * * * ?" />
</bean>
<bean class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
<property name="jobDetail">
<bean parent="methodJobDetail">
<property name="name" value="selectVbapCheck_once"/>
<property name="targetObject" ref="deleteUpdateJobServiceImpl" />
<property name="targetMethod" value="selectVbapCheck" />
</bean>
</property>
<property name="cronExpression" value="* * * * * ?" />
</bean>
</list>
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论