Zpoedit增量

src/main/java/com/huazheng/project/greenplum/service/impl/CheckDeleteServiceImpl.java

@@ -27,6 +27,7 @@ import com.huazheng.project.hana.model.Vbak;
 import com.huazheng.project.hana.model.Vbrk;
 import com.huazheng.project.hana.model.Vbrp;
 import com.huazheng.project.hana.model.Vbuk;
+import com.huazheng.project.hana.model.Zpoedit;
 import com.huazheng.project.hana.model.Zsd06;
 import com.huazheng.project.hana.model.Zsdfhzl;
 
@@ -625,5 +626,34 @@ public class CheckDeleteServiceImpl {
 			redis1Template.opsForValue().set("huazheng:checkDeleteError:Vbuk:rowNum", SomeUtils.getErrorInfoFromException(e));
 		}
 	}
+	public void selectZpoeditCheckByDelete() {
+		try {
+			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
+			opsForValue.setIfAbsent("huazheng:checkDelete:Zpoedit:rowNum", "0");
+			String rowNum = opsForValue.get("huazheng:checkDelete:Zpoedit:rowNum");
+			Zpoedit build = Zpoedit.builder().rowNum(rowNum).build();
+			
+			List<Zpoedit> list = gpMapper.selectZpoeditCheck(build); // 从数仓中查询一组数据
+			if (list.size() == 0) {
+				redis1Template.opsForValue().set("huazheng:checkDelete:Zpoedit:rowNum", "0"); // 计数器复位
+				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
+			}
+			
+			list.forEach(target -> { // 遍历要检查的数据
+				Zpoedit source = sapMapper.selectZpoeditById(target); // 根据主键查询源库中的数据
+				String operator = "none";
+				if (source == null) { // 如果源库中没有数据
+					gpMapper.deleteZpoedit(target); // 删除数仓中的数据
+					operator = "delete";
+				}
+				redis1Template.opsForValue().set("huazheng:checkDelete:Zpoedit:rowNum", target.getRowNum());
+				if (!operator.equals("none")) {
+					log.info(String.format("selectZpoeditcheckDelete --> rowNum:%s, operator:%s", target.getRowNum(), operator));
+				}
+			});
+		} catch (Exception e) {
+			redis1Template.opsForValue().set("huazheng:checkDeleteError:Zpoedit:rowNum", SomeUtils.getErrorInfoFromException(e));
+		}
+	}
 	
 }

src/main/java/com/huazheng/project/greenplum/service/impl/CheckUpdateServiceImpl.java

@@ -28,6 +28,7 @@ import com.huazheng.project.hana.model.Vbak;
 import com.huazheng.project.hana.model.Vbrk;
 import com.huazheng.project.hana.model.Vbrp;
 import com.huazheng.project.hana.model.Vbuk;
+import com.huazheng.project.hana.model.Zpoedit;
 import com.huazheng.project.hana.model.Zsd06;
 import com.huazheng.project.hana.model.Zsdfhzl;
 
@@ -110,6 +111,13 @@ public class CheckUpdateServiceImpl {
 						Afpo tAfpo = gpMapper.selectAfpo(sAfpo); // 查询目标库中afko表
 						cascadeAfpoCheckByUpdate(sAfpo, tAfpo); // 级联更新afko表
 
+						// 级联更新业务
+						List<Zpoedit> sZpoeditList = sapMapper.cascadeZpoeditByAufk(source); // 级联查询源库afko表
+						for (Zpoedit sZpoedit : sZpoeditList) {
+							Zpoedit tZpoedit = gpMapper.selectZpoedit(sZpoedit); // 查询目标库中afko表
+							cascadeZpoeditCheckByUpdate(sZpoedit, tZpoedit); // 级联更新afko表
+						}
+						
 						ThreadUtil.safeSleep(500);
 					}
 				}
@@ -730,6 +738,28 @@ public class CheckUpdateServiceImpl {
 		}
 	}
 	
+	// 所属selectAufkCheckByUpdate的级联
+	private void cascadeZpoeditCheckByUpdate(Zpoedit source, Zpoedit target) {
+		if (target != null) { // 目标库有数据
+			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.updateZpoedit(source); // 更新数据到数仓中
+						break;
+					} catch (RuntimeException e) {
+						log.error(e.getMessage());
+						ThreadUtil.safeSleep(500);
+					}
+				}
+				ThreadUtil.safeSleep(500);
+			}
+		}
+	}
 	// 所属selectLipsCheckUpdate的级联
 	private void cascadeVbukCheckByUpdate(Vbuk source, Vbuk target) {
 		if (target != null) { // 目标库有数据

src/main/java/com/huazheng/project/hana/mapper/SapMapper.java

@@ -110,6 +110,7 @@ public interface SapMapper {
 	public List<Konv> cascadeKonvByVbak(Vbak vbak);
 	public List<Makt> cascadeMaktByMara(Mara mara);
 	public List<Vbuk> cascadeVbukByLips(Lips lips);
+	public List<Zpoedit> cascadeZpoeditByAufk(Aufk aufk);
 	
 	// ......
 	public Bkpf selectBkpfById(Bkpf target);

src/main/resources/mapper/hana/SapMapper_hana.xml

@@ -477,6 +477,12 @@
     	from ${hana_user}.vbuk
     	where vbeln = #{vbeln} ${hana_mandt}
     </select>
+    <select id="cascadeZpoeditByAufk" parameterType="Aufk" resultType="Zpoedit">
+    	select 
+       		mandt, pwerk, aufnr, reason
+    	from ${hana_user}.Zpo_edit
+    	where aufnr = #{aufnr} ${hana_mandt}
+    </select>
     
     
     

src/main/resources/spring-init.xml

@@ -978,6 +978,15 @@
 					</property>
 					<property name="cronExpression" value="* * * * * ?" />
 				</bean>
+				<bean class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
+					<property name="jobDetail">
+						<bean parent="methodJobDetail">
+							<property name="targetObject" ref="checkDeleteServiceImpl" />
+							<property name="targetMethod" value="selectZpoeditCheckByDelete" />
+						</bean>
+					</property>
+					<property name="cronExpression" value="* * * * * ?" />
+				</bean>
 				
 				
 			</list>