package com.huazheng.project.greenplum.service.impl;

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Service;

import com.huazheng.project.greenplum.mapper.GPMapper;
import com.huazheng.project.greenplum.utils.SomeUtils;
import com.huazheng.project.hana.mapper.SapMapper;
import com.huazheng.project.hana.model.Afko;
import com.huazheng.project.hana.model.Afpo;
import com.huazheng.project.hana.model.Aufk;
import com.huazheng.project.hana.model.Ausp;
import com.huazheng.project.hana.model.Bkpf;
import com.huazheng.project.hana.model.Bsad;
import com.huazheng.project.hana.model.Bsid;
import com.huazheng.project.hana.model.Kna1;
import com.huazheng.project.hana.model.Knvv;
import com.huazheng.project.hana.model.Konv;
import com.huazheng.project.hana.model.Likp;
import com.huazheng.project.hana.model.Lips;
import com.huazheng.project.hana.model.Makt;
import com.huazheng.project.hana.model.Mara;
import com.huazheng.project.hana.model.Pa0002;
import com.huazheng.project.hana.model.T023t;
import com.huazheng.project.hana.model.Vbak;
import com.huazheng.project.hana.model.Vbap;
import com.huazheng.project.hana.model.VbapAdv;
import com.huazheng.project.hana.model.Vbep;
import com.huazheng.project.hana.model.Vbkd;
import com.huazheng.project.hana.model.Vbpa;
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.Vbup;
import com.huazheng.project.hana.model.Zmdpc;
import com.huazheng.project.hana.model.Zpoedit;
import com.huazheng.project.hana.model.Zsd06;
import com.huazheng.project.hana.model.Zsdfhzl;
import com.huazheng.project.mssql.mapper.CrmMapper;
import com.huazheng.project.mssql.model.SalesContractProcess;
import com.huazheng.project.mssql.model.SalesContractProcessMX;
import com.huazheng.project.mssql.model.SalesForecastProcess;
import com.huazheng.project.mssql.model.SalesForecastProcessMX;
import com.huazheng.project.mssql.model.SalesInquiryProcess;
import com.huazheng.project.mssql.model.SalesInquiryProcessMX;
import com.huazheng.project.mssql.model.SampleApplicationProcess;
import com.huazheng.project.mssql.model.SampleApplicationProcessMingXi;
import com.huazheng.project.mssql.model.SampleClosingProcess;
import com.huazheng.project.mssql.model.SampleClosingProcessMingXi;
import com.huazheng.project.mssql2.mapper.HzcrmMapper;
import com.huazheng.project.mssql2.model.SysSAPreturnNo;
import com.huazheng.project.mssql4.mapper.UltimusDBMapper;
import com.huazheng.project.mssql4.model.Tasks;
import com.huazheng.project.mysql.mapper.TmsMapper;
import com.huazheng.project.mysql.model.Handover;
import com.huazheng.project.mysql.model.HandoverTask;
import com.huazheng.project.mysql.model.TransformHistoryNode;
import com.huazheng.project.mysql.model.TransformNewNode;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.json.JSONUtil;
import lombok.extern.log4j.Log4j2;

@Log4j2
@Service
public class CheckUpdateServiceImpl {

	@Autowired
	private RedisTemplate<String, String> redis1Template;
	@Autowired
	private GPMapper gpMapper;
	@Autowired
	private SapMapper sapMapper;
	@Autowired
	private CrmMapper crmMapper;
	@Autowired
	private TmsMapper tmsMapper;
	@Autowired
	private HzcrmMapper hzcrmMapper;
	@Autowired
	private UltimusDBMapper ultimusDBMapper;
	
	private ExecutorService service = Executors.newFixedThreadPool(15);
	
	
	public void checkJob6() {
		selectAufkCheckByUpdate();
		selectBkpfCheckByUpdate();
		selectKna1CheckByUpdate();
		selectPa0002CheckUpdate();
		selectZsd06CheckUpdate();
		selectZsdfhzlCheckUpdate();
		selectMaraCheckUpdate();
		selectSalesForecastProcessCheckUpdate();
		selectHandoverCheckUpdate();
		selectHandoverTaskCheckUpdate();
		selectTransformHistoryNodeCheckUpdate();
		selectTransformNewNodeCheckUpdate();
		selectVbapCheckUpdate();
	}
	
	public void selectAufkCheckByUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Aufk:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Aufk:rowids"));
			Aufk build = Aufk.builder().rowids(rowids).build();

			List<Aufk> slist = sapMapper.selectAufkCheckByUpdate(build); // 从源库中按更新时间查询,只更新今天的数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:Aufk:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历源库中的数据，去目标库中查询数据
				Aufk target = gpMapper.selectAufk(source);
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getErdat() != null && source.getErfzeit() != null) {
							String erdat2 = new StringBuffer(source.getErdat()).insert(4, "-").insert(7, "-").toString();
							String erzet2 = new StringBuffer(source.getErfzeit()).insert(2, ":").insert(5, ":").toString();
							String dateStr = erdat2 + " " + erzet2;
							Date date = DateUtil.parse(dateStr);
							source.setErdat1(date);
							source.setErdat2(date);
						}
						String aedat = source.getAedat();
						if ("00000000".equals(aedat)) {
							source.setAedat(null);
						} else {
							Date date = DateUtil.parse(aedat, "yyyyMMdd");
							String format = DateUtil.format(date, "yyyy-MM-dd");
							source.setAedat(format);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateAufk(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}

						// 级联更新业务
						Afko sAfko = sapMapper.cascadeAfkoByAufk(source); // 级联查询源库afko表
						Afko tAfko = gpMapper.selectAfko(sAfko); // 查询目标库中afko表
						cascadeAfkoCheckByUpdate(sAfko, tAfko); // 级联更新afko表

						// 级联更新业务
						Afpo sAfpo = sapMapper.cascadeAfpoByAufk(source); // 级联查询源库afko表
						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);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Aufk:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectAufkcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Aufk:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectBkpfCheckByUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Bkpf:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Bkpf:rowids"));
			Bkpf build = Bkpf.builder().rowids(rowids).build();

			List<Bkpf> slist = sapMapper.selectBkpfCheckByUpdate(build); // 从源库中按更新时间查询,只更新今天的数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:Bkpf:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历源库中的数据，去目标库中查询数据
				Bkpf target = gpMapper.selectBkpf(source);
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						String aedat = source.getAedat();
						if ("00000000".equals(aedat)) {
							source.setAedat(null);
						} else {
							Date date = DateUtil.parse(aedat, "yyyyMMdd");
							String format = DateUtil.format(date, "yyyy-MM-dd");
							source.setAedat(format);
						}
						String cpudt = source.getCpudt();
						String cputm = source.getCputm();
						if ("00000000".equals(cpudt)) {
							source.setCpudt_cputm(null);
						} else {
							Date date = DateUtil.parse(cpudt + cputm, "yyyyMMddHHmmss");
							source.setCpudt_cputm(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateBkpf(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<Bsad> sBsadList = sapMapper.cascadeBsadByBkpf(source); // 级联查询源库afko表
						for (Bsad sBsad : sBsadList) {
							Bsad tBsad = gpMapper.selectBsad(sBsad); // 查询目标库中afko表
							cascadeBsadCheckByUpdate(sBsad, tBsad); // 级联更新afko表
						}
						
						// 级联更新业务
						List<Bsid> sBsidList = sapMapper.cascadeBsidByBkpf(source); // 级联查询源库afko表
						for (Bsid sBsid : sBsidList) {
							Bsid tBsid = gpMapper.selectBsid(sBsid); // 查询目标库中afko表
							cascadeBsidCheckByUpdate(sBsid, tBsid); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:Bkpf:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectBkpfcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Bkpf:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectKna1CheckByUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Kna1:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Kna1:rowids"));
			Kna1 build = Kna1.builder().rowids(rowids).build();

			List<Kna1> slist = sapMapper.selectKna1CheckUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:Kna1:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据，去目标库中查询数据
				Kna1 target = gpMapper.selectKna1(source);
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						String updat = source.getUpdat();
						if ("00000000".equals(updat)) {
							source.setUpdat(null);
						} else {
							Date date = DateUtil.parse(updat, "yyyyMMdd");
							String format = DateUtil.format(date, "yyyy-MM-dd");
							source.setUpdat(format);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateKna1(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}

						// 级联更新业务
						List<Knvv> sKnvvList = sapMapper.cascadeKnvvByKna1(source); // 级联查询源库afko表
						for (Knvv sKnvv : sKnvvList) {
							Knvv tKnvv = gpMapper.selectKnvv(sKnvv); // 查询目标库中afko表
							cascadeKnvvCheckByUpdate(sKnvv, tKnvv); // 级联更新afko表
						}

						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Kna1:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectKna1checkUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Kna1:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	public void selectPa0002CheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Pa0002:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Pa0002:rowids"));
			Pa0002 build = Pa0002.builder().rowids(rowids).build();

			List<Pa0002> slist = sapMapper.selectPa0002CheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Pa0002:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Pa0002 target = gpMapper.selectPa0002(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getEndda() != null) {
							String erdat2 = new StringBuffer(source.getEndda()).insert(4, "-").insert(7, "-").toString();
							Date date = DateUtil.parse(erdat2);
							source.setEndda1(date);
						}
						source.setYeWuYMC(source.getNachn() + source.getVorna());
						// ===============================
						while (true) {
							try {
								gpMapper.updatePa0002(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Pa0002:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectPa0002checkUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Pa0002:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	public void selectZsd06CheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Zsd06:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Zsd06:rowids"));
			Zsd06 build = Zsd06.builder().rowids(rowids).build();

			List<Zsd06> slist = sapMapper.selectZsd06CheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Zsd06:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Zsd06 target = gpMapper.selectZsd06(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getDatum() != null) {
							String erdat2 = new StringBuffer(source.getDatum()).insert(4, "-").insert(7, "-").toString();
							Date date = DateUtil.parse(erdat2);
							source.setDatum1(date);
						}
						// source.setCrt_datum1(SomeUtils.caDate(source.getCrt_datum()));
						// source.setCha_datum1(SomeUtils.caDate(source.getCha_datum()));
						String crtdatum = source.getCrt_datum();
						String crttime = source.getCrt_time(); 
						if ("00000000".equals(crtdatum)) {
							source.setCrt_datum1(null);
						} else {
							Date date = DateUtil.parse(crtdatum+crttime, "yyyyMMddHHmmss");
							source.setCrt_datum1(date);
						}
						
						String chadatum = source.getCha_datum();
						String chatime = source.getCha_time(); 
						if ("00000000".equals(chadatum)) {
							source.setCha_datum1(null);
						} else {
							Date date = DateUtil.parse(chadatum+chatime, "yyyyMMddHHmmss");
							source.setCha_datum1(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateZsd06(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Zsd06:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectZsd06checkUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Zsd06:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectZsdfhzlCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Zsdfhzl:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Zsdfhzl:rowids"));
			Zsdfhzl build = Zsdfhzl.builder().rowids(rowids).build();

			List<Zsdfhzl> slist = sapMapper.selectZsdfhzlCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Zsdfhzl:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Zsdfhzl target = gpMapper.selectZsdfhzl(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getCdate() != null && source.getCtime() != null) {
							String erdat2 = new StringBuffer(source.getCdate()).insert(4, "-").insert(7, "-").toString();
							String erzet2 = new StringBuffer(source.getCtime()).insert(2, ":").insert(5, ":").toString();
							String dateStr = erdat2 + " " + erzet2;
							Date date = DateUtil.parse(dateStr);
							source.setCdate1(date);
						}
						if (source.getFh_date() != null && source.getFh_date().indexOf("--") == -1) {
							String erdat2 = new StringBuffer(source.getFh_date()).insert(4, "-").insert(7, "-").toString();
							Date date = DateUtil.parse(erdat2);
							source.setFh_date1(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateZsdfhzl(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:Zsdfhzl:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectZsdfhzlcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Zsdfhzl:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectMaraCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Mara:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Mara:rowids"));
			Mara build = Mara.builder().rowids(rowids).build();

			List<Mara> slist = sapMapper.selectMaraCheckByUpdate(build); // 从源库中按更新时间查询,只更新今天的数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:Mara:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Mara target = gpMapper.selectMara(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getErsda() != null) {
							String erdat2 = new StringBuffer(source.getErsda()).insert(4, "-").insert(7, "-").toString();
							Date date = DateUtil.parse(erdat2);
							source.setErsda1(date);
						}
						if (source.getLaeda() != null) {
							String erdat2 = new StringBuffer(source.getLaeda()).insert(4, "-").insert(7, "-").toString();
							Date date = DateUtil.parse(erdat2);
							source.setLaeda1(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateMara(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}

						// 级联更新业务
						List<Ausp> sAuspList = sapMapper.cascadeAuspByMara(source); // 级联查询源库afko表
						for (Ausp sAusp : sAuspList) {
							Ausp tAusp = gpMapper.selectAusp(sAusp); // 查询目标库中afko表
							cascadeAuspCheckByUpdate(sAusp, tAusp); // 级联更新afko表
						}

						List<Makt> sMaktList = sapMapper.cascadeMaktByMara(source); // 级联查询源库afko表
						for (Makt sMakt : sMaktList) {
							Makt tMakt = gpMapper.selectMakt(sMakt); // 查询目标库中afko表
							cascadeMaktCheckByUpdate(sMakt, tMakt); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:Mara:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectMaracheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Mara:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	// 所属selectBkpfCheckByUpdate的级联
	private void cascadeBsadCheckByUpdate(Bsad source, Bsad 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);
				// ===============================
				if (StrUtil.isBlank(source.getUmsks())) {
					source.setUmskz(""); // 属于主键组，不能为空
				}
				String augdts = source.getAugdts(); // 日期转义
				if (augdts == null || augdts.equals("00000000") || StrUtil.isBlank(augdts)) {
					Date date = DateUtil.parse("1970-01-01 00:00:00");
					source.setAugdt(date);
				} else {
					String erdat2 = new StringBuffer(augdts).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setAugdt(date);
				}
				String zfbdts = source.getZfbdts(); // 转义日期
				if (zfbdts == null || zfbdts.equals("00000000") || StrUtil.isBlank(zfbdts)) {
					Date date = DateUtil.parse("1970-01-01 00:00:00");
					source.setZfbdt(date);
				} else {
					String erdat2 = new StringBuffer(zfbdts).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setZfbdt(date);
				}
				String budats = source.getBudats(); // 转义日期
				if (budats == null || budats.equals("00000000") || StrUtil.isBlank(budats)) {
					Date date = DateUtil.parse("1970-01-01 00:00:00");
					source.setBudat(date);
					source.setBudatc("1970-01");
					source.setDqsjc(null); // 到期时间不用计算
				} else {
					String erdat2 = new StringBuffer(budats).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setBudat(date);
					source.setBudatc(DateUtil.format(date, "yyyy-MM"));
					
					int budatDay = Integer.parseInt(DateUtil.format(source.getBudat(), "dd")); // 过账日期
					if (budatDay > 26) { // budat(过账日期) > 26号 ：zfbdt(基准日期)+2个月
						source.setDqsjc(DateUtil.offset(source.getZfbdt(), DateField.MONTH, 2)); // 到期时间
						// 根据过账日期判断日是否>26，大于的话，就在过账日期的基础上加一个月，小于等于的话就是不变
						source.setBudatd(DateUtil.format(DateUtil.offset(source.getBudat(), DateField.MONTH, 1), "yyyy-MM"));
					} else if (budatDay <= 26) { // budat(过账日期) <= 26号 ：zfbdt(基准日期)+1个月
						source.setDqsjc(DateUtil.offset(source.getZfbdt(), DateField.MONTH, 1)); // 到期时间
						// 根据过账日期判断日是否>26，大于的话，就在过账日期的基础上加一个月，小于等于的话就是不变
						source.setBudatd(DateUtil.format(source.getBudat(), "yyyy-MM"));
					}
					
					if ("Z001".equals(source.getZterm())) {
						source.setDqsjc(source.getZfbdt()); // 现款，到期时间=基准日期
					}
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateBsad(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectBkpfCheckByUpdate的级联
	private void cascadeBsidCheckByUpdate(Bsid source, Bsid 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);
				// ===============================
				if (StrUtil.isBlank(source.getUmsks())) {
					source.setUmskz(""); // 属于主键组，不能为空
				}
				String augdts = source.getAugdts(); // 日期转义
				if (augdts == null || augdts.equals("00000000") || StrUtil.isBlank(augdts)) {
					Date date = DateUtil.parse("1970-01-01 00:00:00");
					source.setAugdt(date);
				} else {
					String erdat2 = new StringBuffer(augdts).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setAugdt(date);
				}
				String zfbdts = source.getZfbdts(); // 转义日期
				if (zfbdts == null || zfbdts.equals("00000000") || StrUtil.isBlank(zfbdts)) {
					Date date = DateUtil.parse("1970-01-01 00:00:00");
					source.setZfbdt(date);
				} else {
					String erdat2 = new StringBuffer(zfbdts).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setZfbdt(date);
				}
				String budats = source.getBudats(); // 转义日期
				if (budats == null || budats.equals("00000000") || StrUtil.isBlank(budats)) {
					Date date = DateUtil.parse("1970-01-01 00:00:00");
					source.setBudat(date);
					source.setBudatc("1970-01");
					source.setDqsjc(null); // 到期时间不用计算
				} else {
					String erdat2 = new StringBuffer(budats).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setBudat(date);
					source.setBudatc(DateUtil.format(date, "yyyy-MM"));
					
					int budatDay = Integer.parseInt(DateUtil.format(source.getBudat(), "dd")); // 过账日期
					if (budatDay > 26) { // budat(过账日期) > 26号 ：zfbdt(基准日期)+2个月
						source.setDqsjc(DateUtil.offset(source.getZfbdt(), DateField.MONTH, 2)); // 到期时间
						// 根据过账日期判断日是否>26，大于的话，就在过账日期的基础上加一个月，小于等于的话就是不变
						source.setBudatd(DateUtil.format(DateUtil.offset(source.getBudat(), DateField.MONTH, 1), "yyyy-MM"));
					} else if (budatDay <= 26) { // budat(过账日期) <= 26号 ：zfbdt(基准日期)+1个月
						source.setDqsjc(DateUtil.offset(source.getZfbdt(), DateField.MONTH, 1)); // 到期时间
						// 根据过账日期判断日是否>26，大于的话，就在过账日期的基础上加一个月，小于等于的话就是不变
						source.setBudatd(DateUtil.format(source.getBudat(), "yyyy-MM"));
					}
					
					if ("Z001".equals(source.getZterm())) {
						source.setDqsjc(source.getZfbdt()); // 现款，到期时间=基准日期
					}
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateBsid(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属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) { // 目标库有数据
			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.updateVbuk(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	
	// 所属selectLipsCheckUpdate的级联
	private void cascadeVbupCheckByUpdate(Vbup source, Vbup 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.updateVbup(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	
	// 所属selectVbapCheckUpdate的级联
	private void cascadeVbkdCheckByUpdate(Vbkd source, Vbkd 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.updateVbkd(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	
	// 所属selectMaraCheckUpdate的级联
	private void cascadeMaktCheckByUpdate(Makt source, Makt 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);
				// ===============================
				String maktx = source.getMaktx();
				if (maktx != null && !maktx.equals("")) {
					String[] split = maktx.split("-");
					for (int i = split.length - 1; i >= 0; i--) {
						String string = split[i];
						if (string.indexOf("*") != -1) {
							source.setMaktxSize(string.split("\\*")[0]); // 分割出尺寸数据
							break;
						}
					}
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateMakt(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectMaraCheckUpdate的级联
	private void cascadeAuspCheckByUpdate(Ausp source, Ausp 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.updateAusp(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectKna1CheckByUpdate的级联
	public void cascadeKnvvCheckByUpdate(Knvv source, Knvv 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.updateKnvv(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectAufkCheckByUpdate的级联
	private void cascadeAfkoCheckByUpdate(Afko source, Afko 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);
				// ===============================
				if (source.getGltrp() != null) {
					String erdat2 = new StringBuffer(source.getGltrp()).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setGltrp1(date);
				}
				if (source.getGstrp() != null) {
					String erdat2 = new StringBuffer(source.getGstrp()).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setGstrp1(date);
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateAfko(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectAufkCheckByUpdate的级联
	private void cascadeAfpoCheckByUpdate(Afpo source, Afpo 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.updateAfpo(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectVbrkCheckUpdate的级联
	private void cascadeVbrpCheckByUpdate(Vbrp source, Vbrp 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.updateVbrp(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectVbakCheckUpdate的级联
	private void cascadeKonvCheckByUpdate(Konv source, Konv 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.updateKonv(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	
	
	public void selectSalesForecastProcessCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:SalesForecastProcess:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:SalesForecastProcess:rowids");
			SalesForecastProcess build = SalesForecastProcess.builder().id(rowids).build();
			
			List<SalesForecastProcess> slist = crmMapper.selectSalesForecastProcessCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:SalesForecastProcess:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				SalesForecastProcess target = gpMapper.selectSalesForecastProcess(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId();
				if (target != null) {
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						String kaiShiSJs = source.getKaiShiSJs(); // 时间数据纠错
						if (kaiShiSJs == null || kaiShiSJs.equals("00000000") || StrUtil.isBlank(kaiShiSJs)) {
							source.setKaiShiSJ(null);
						} else {
							if (kaiShiSJs.indexOf("--") == -1) {
								Date date = DateUtil.parse(kaiShiSJs);
								source.setKaiShiSJ(date);
							}
						}
						String jieShuSJs = source.getJieShuSJs(); // 时间数据纠错
						if (jieShuSJs == null || jieShuSJs.equals("00000000") || StrUtil.isBlank(jieShuSJs)) {
							source.setJieShuSJ(null);
						} else {
							Date date = DateUtil.parse(jieShuSJs);
							source.setJieShuSJ(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateSalesForecastProcess(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<SalesForecastProcessMX> sSalesForecastProcessMXList = crmMapper.cascadeSalesForecastProcessMXBySalesForecastProcess(source); // 级联查询源库afko表
						for (SalesForecastProcessMX sSalesForecastProcessMX : sSalesForecastProcessMXList) {
							SalesForecastProcessMX tSalesForecastProcessMX = gpMapper.selectSalesForecastProcessMX(sSalesForecastProcessMX); // 查询目标库中afko表
							cascadeSalesForecastProcessMXCheckByUpdate(sSalesForecastProcessMX, tSalesForecastProcessMX); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:SalesForecastProcess:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectSalesForecastProcesscheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:SalesForecastProcess:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	
	// 所属selectSalesContractProcessCheckUpdate的级联
	private void cascadeSalesContractProcessMXCheckByUpdate(SalesContractProcessMX source, SalesContractProcessMX 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);
				// ===============================
				Object chengben = source.getChengben();
				if (chengben != null && chengben.toString().equals("")) {
					source.setChengben(null);
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateSalesContractProcessMX(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectSalesForecastProcessCheckUpdate的级联
	private void cascadeSalesForecastProcessMXCheckByUpdate(SalesForecastProcessMX source, SalesForecastProcessMX 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);
				// ===============================
				if (source.getJiHuaJDL() != null) {
					source.setJiHuaJDL(ReUtil.getFirstNumber(source.getJiHuaJDL()) + "");
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateSalesForecastProcessMX(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectSampleApplicationProcessCheckUpdate的级联
	private void cascadeSampleApplicationProcessMingXiCheckByUpdate(SampleApplicationProcessMingXi source, SampleApplicationProcessMingXi 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);
				// ===============================
				String hangXiangMH = source.getHangXiangMH();
				if (StrUtil.isBlank(hangXiangMH)) {
					source.setHangxiangmh1(null);
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateSampleApplicationProcessMingXi(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectSampleClosingProcessCheckUpdate的级联
	private void cascadeSampleClosingProcessMingXiCheckByUpdate(SampleClosingProcessMingXi source, SampleClosingProcessMingXi 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.updateSampleClosingProcessMingXi(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectSalesInquiryProcessCheckUpdate的级联
	private void cascadeSalesInquiryProcessMXCheckByUpdate(SalesInquiryProcessMX source, SalesInquiryProcessMX 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.updateSalesInquiryProcessMX(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	public void selectHandoverCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Handover:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:Handover:rowids");
			Handover build = Handover.builder().handover_id(rowids).build();
			
			List<Handover> slist = tmsMapper.selectHandoverCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:Handover:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				Handover target = gpMapper.selectHandover(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getHandover_id();
				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.updateHandover(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:Handover:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectHandovercheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Handover:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectHandoverTaskCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:HandoverTask:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:HandoverTask:rowids");
			HandoverTask build = HandoverTask.builder().task_id(rowids).build();
			
			List<HandoverTask> slist = tmsMapper.selectHandoverTaskCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:HandoverTask:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				HandoverTask target = gpMapper.selectHandoverTask(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getTask_id();
				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.updateHandoverTask(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:HandoverTask:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectHandoverTaskcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:HandoverTask:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectTransformHistoryNodeCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:TransformHistoryNode:rowids", "0");
			Integer rowids = Integer.parseInt(opsForValue.get("huazheng:checkUpdate:TransformHistoryNode:rowids"));
			TransformHistoryNode build = TransformHistoryNode.builder().id(rowids).build();
			// 新加了字段要调试
			List<TransformHistoryNode> slist = tmsMapper.selectTransformHistoryNodeCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:TransformHistoryNode:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				TransformHistoryNode target = gpMapper.selectTransformHistoryNode(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId().toString();
				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.updateTransformHistoryNode(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:TransformHistoryNode:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectTransformHistoryNodecheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:TransformHistoryNode:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectTransformNewNodeCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:TransformNewNode:rowids", "0");
			Integer rowids = Integer.parseInt(opsForValue.get("huazheng:checkUpdate:TransformNewNode:rowids"));
			TransformNewNode build = TransformNewNode.builder().id(rowids).build();
			
			List<TransformNewNode> slist = tmsMapper.selectTransformNewNodeCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:TransformNewNode:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				TransformNewNode target = gpMapper.selectTransformNewNode(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId().toString();
				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.updateTransformNewNode(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:TransformNewNode:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectTransformNewNodecheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:TransformNewNode:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	// 独立的
	public void selectVbapCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Vbap:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Vbap:rowids"));
			Vbap build = Vbap.builder().rowids(rowids).build();

			List<Vbap> slist = sapMapper.selectVbapCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Vbap:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				if (source.getErdat() != null && source.getErzet() != null) {
					String erdat2 = new StringBuffer(source.getErdat()).insert(4, "-").insert(7, "-").toString();
					String erzet2 = new StringBuffer(source.getErzet()).insert(2, ":").insert(5, ":").toString();
					String dateStr = erdat2 + " " + erzet2;
					Date date = DateUtil.parse(dateStr);
					source.setErdat1(date);
					source.setErdat2(date);
				}
				Vbap target = gpMapper.selectVbap(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(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.updateVbap(source); // 更新数据到数仓中
								gpMapper.updateVbapAdv(source);
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<Vbep> sVbepList = sapMapper.cascadeVbepByVbap(source); // 级联查询源库Vbep表
						for (Vbep sVbep : sVbepList) {
							Vbep tVbep = gpMapper.selectVbep(sVbep); // 查询目标库中Vbep表
							cascadeVbepCheckByUpdate(sVbep, tVbep); // 级联更新afko表
						}
						// 级联更新业务
						List<Vbpa> sVbpaList = sapMapper.cascadeVbpaByVbap(source); // 级联查询源库Vbpa表
						for (Vbpa sVbpa : sVbpaList) {
							Vbpa tVbpa = gpMapper.selectVbpa(sVbpa); // 查询目标库中Vbpa表
							cascadeVbpaCheckByUpdate(sVbpa, tVbpa); // 级联更新afko表
						}
						// 级联更新业务
						List<Zmdpc> sZmdpcList = sapMapper.cascadeZmdpcByVbap(source); // 级联查询源库Zmdpc表
						for (Zmdpc sZmdpc : sZmdpcList) {
							Zmdpc tZmdpc = gpMapper.selectZmdpc(sZmdpc); // 查询目标库中Zmdpc表
							cascadeZmdpcCheckByUpdate(sZmdpc, tZmdpc); // 级联更新afko表
						}
						// 级联更新业务
						List<SysSAPreturnNo> sSysSAPreturnNoList = hzcrmMapper.cascadeSysSAPreturnNoByVbap(source); // 级联查询源库SysSAPreturnNo表
						for (SysSAPreturnNo sSysSAPreturnNo : sSysSAPreturnNoList) {
							SysSAPreturnNo tSysSAPreturnNo = gpMapper.selectSysSAPreturnNo(sSysSAPreturnNo); // 查询目标库中SysSAPreturnNo表
							cascadeSysSAPreturnNoCheckByUpdate(sSysSAPreturnNo, tSysSAPreturnNo); // 级联更新afko表
						}
						// 级联更新业务
						List<Vbkd> sVbkdList = sapMapper.cascadeVbkdByVbap(source); // 级联查询源库Vbkd表
						for (Vbkd sVbkd : sVbkdList) {
							Vbkd tVbkd = gpMapper.selectVbkd(sVbkd); // 查询目标库中Vbkd表
							cascadeVbkdCheckByUpdate(sVbkd, tVbkd); // 级联更新Vbkd表
						}
						
						ThreadUtil.safeSleep(500);
						try {
							Future<VbapAdv> vbapAdvSubmit = service.submit(() -> {
								return gpMapper.selectVbapAdv(source);
							});
							
							Future<BigDecimal> ccslSubmit = service.submit(() -> {
								return gpMapper.caLips4ccsl(source);
							});
							Future<BigDecimal> jhslSubmit = service.submit(() -> {
								return gpMapper.caVbep4jhsl(source);
							});
							Future<BigDecimal> rkslSubmit = service.submit(() -> {
								return gpMapper.caAfpo4rksl(source);
							});
							Future<BigDecimal> pcslSubmit = service.submit(() -> {
								return gpMapper.caAfpo4pcsl(source);
							});
							Future<BigDecimal> kpslSubmit = service.submit(() -> {
								return gpMapper.caVbrp4kpsl(source);
							});
							Future<Date> jhsjSubmit = service.submit(() -> {
								return gpMapper.caVbep4jhsj(source);
							});
							Future<Date> kpsjSubmit = service.submit(() -> {
								return gpMapper.caVbrk4kpsj(source);
							});
							Future<Date> rksjSubmit = service.submit(() -> {
								return gpMapper.caAufm4rksj(source);
							});
							Future<Date> ccsjSubmit = service.submit(() -> {
								return gpMapper.caHandover4ccsj(source);
							});
							Future<Date> wlsjSubmit = service.submit(() -> {
								return gpMapper.caHandover4wlsj(source);
							});
							
							VbapAdv vbapAdv = vbapAdvSubmit.get();
							vbapAdv.setCcsl(ccslSubmit.get());
							vbapAdv.setJhsl(jhslSubmit.get());
							vbapAdv.setRksl(rkslSubmit.get());
							vbapAdv.setPcsl(pcslSubmit.get());
							vbapAdv.setKpsl(kpslSubmit.get());
							vbapAdv.setJhsj(jhsjSubmit.get());
							vbapAdv.setKpsj(kpsjSubmit.get());
							vbapAdv.setRksj(rksjSubmit.get());
							vbapAdv.setCcsj(ccsjSubmit.get());
							vbapAdv.setWlsj(wlsjSubmit.get());
							vbapAdv.setHtsl(source.getKwmeng());
							vbapAdv.setPssj(source.getErdat1());
							vbapAdv.setPcsj(source.getErdat1());
							gpMapper.updatevbapAdv4CaAll(vbapAdv);
							
							Aufk aufk = gpMapper.selectVbapAdv4Aufk(source); // vbap-aufk
							if (aufk != null) {
								aufk.setVbap(source);
								gpMapper.updateVbapAdv4Aufk(aufk); // vbap-aufk
							}

							SysSAPreturnNo sysSAPreturnNo = gpMapper.selectVbapAdv4SysSAPreturnNo(source); // vbap-SysSAPreturnNo
							if (sysSAPreturnNo != null) {
								sysSAPreturnNo.setVbap(source);
								gpMapper.updateVbapAdv4SysSAPreturnNo(sysSAPreturnNo); // vbap-SysSAPreturnNo
							}

							SalesContractProcess salesContractProcess = gpMapper.selectVbapAdv4SalesContractProcess(source); // vbap-SysSAPreturnNo-SalesContractProcess
							if (salesContractProcess != null) {
								salesContractProcess.setVbap(source);
								gpMapper.updateVbapAdv4SalesContractProcess(salesContractProcess); // vbap-SysSAPreturnNo-SalesContractProcess
							}

							Mara mara = gpMapper.selectVbapAdv4Mara(source); // vbap-Mara
							if (mara != null) {
								mara.setVbap(source);
								gpMapper.updateVbapAdv4Mara(mara); // vbap-Mara
							}

							Makt makt = gpMapper.selectVbapAdv4Makt(source); // vbap-Makt
							if (makt != null) {
								makt.setVbap(source);
								gpMapper.updateVbapAdv4Makt(makt); // vbap-Makt
							}

							Zsd06 zsd06 = gpMapper.selectVbapAdv4Zsd06(source); // vbap-Zsd06
							if (zsd06 != null) {
								zsd06.setVbap(source);
								gpMapper.updateVbapAdv4Zsd06(zsd06); // vbap-Zsd06
							}

							Vbak vbak = gpMapper.selectVbapAdv4Vbak(source); // vbap-Vbak
							if (vbak != null) {
								vbak.setVbap(source);
								gpMapper.updateVbapAdv4Vbak(vbak); // vbap-Vbak
							}

							Likp likp = gpMapper.selectVbapAdv4Likp(source); // vbap-lips-Likp
							if (likp != null) {
								likp.setVbap(source);
								gpMapper.updateVbapAdv4Likp(likp); // vbap-lips-Likp
							}

							Handover handover = gpMapper.selectVbapAdv4Handover(source); // vbap-lips-Likp-Handover
							if (handover != null) {
								handover.setVbap(source);
								gpMapper.updateVbapAdv4Handover(handover); // vbap-lips-Likp-Handover
							}

							Zmdpc zmdpc = gpMapper.selectVbapAdv4Zmdpc(source); // vbap-Zmdpc
							if (zmdpc != null) {
								zmdpc.setVbap(source);
								gpMapper.updateVbapAdv4Zmdpc(zmdpc); // vbap-Zmdpc
							}

							T023t t023t = gpMapper.selectVbapAdv4T023t(source); // vbap-mara-T023t
							if (t023t != null) {
								t023t.setVbap(source);
								gpMapper.updateVbapAdv4T023t(t023t); // vbap-mara-T023t
							}
							
							Kna1 kna1 = gpMapper.selectVbapAdv4Kna1(source); // vbap-vbak-Kna1
							if (kna1 != null) {
								kna1.setVbap(source);
								gpMapper.updateVbapAdv4Kna1(kna1); // vbap-vbak-Kna1
							}
							
						} catch (Exception e) {
							redis1Template.opsForValue().set("huazheng:checkUpdateError:Vbap:rowids", SomeUtils.getErrorInfoFromException(e));
						}
						
					}
				}
				
				redis1Template.opsForValue().set("huazheng:checkUpdate:Vbap:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectVbapcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Vbap:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	// 所属selectVbapCheckUpdate的级联
	private void cascadeVbepCheckByUpdate(Vbep source, Vbep 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);
				// ===============================
				if (source.getEdatu() != null) {
					String erdat2 = new StringBuffer(source.getEdatu()).insert(4, "-").insert(7, "-").toString();
					Date date = DateUtil.parse(erdat2);
					source.setEdatu1(date);
				}
				// ===============================
				while (true) {
					try {
						gpMapper.updateVbep(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectVbapCheckUpdate的级联
	private void cascadeVbpaCheckByUpdate(Vbpa source, Vbpa 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.updateVbpa(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectVbapCheckUpdate的级联
	private void cascadeZmdpcCheckByUpdate(Zmdpc source, Zmdpc 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.updateZmdpc(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	// 所属selectVbapCheckUpdate的级联
	private void cascadeSysSAPreturnNoCheckByUpdate(SysSAPreturnNo source, SysSAPreturnNo 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.updateSysSAPreturnNo(source); // 更新数据到数仓中
						break;
					} catch (RuntimeException e) {
						log.error(e.getMessage());
						ThreadUtil.safeSleep(500);
					}
				}
				ThreadUtil.safeSleep(500);
			}
		}
	}
	
	// 独立的
	public void selectLikpCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Likp:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Likp:rowids"));
			Likp build = Likp.builder().rowids(rowids).build();

			List<Likp> slist = sapMapper.selectLikpCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Likp:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Likp target = gpMapper.selectLikp(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						String wadat_ist = source.getWadat_ist();
						String lfuhr = source.getLfuhr();
						if ("00000000".equals(wadat_ist)) {
							source.setWadat_ist2lfuhr(null);
						} else {
							Date date = DateUtil.parse(wadat_ist + lfuhr, "yyyyMMddHHmmss");
							source.setWadat_ist2lfuhr(date);
						}
						source.setErdat1(SomeUtils.caDate(source.getErdat())); // 日期00000000格式转换,已处理异常
						source.setWadat_ist1(SomeUtils.caDate(source.getWadat_ist())); // 日期00000000格式转换,已处理异常
						// ===============================
						while (true) {
							try {
								gpMapper.updateLikp(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Likp:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectLikpcheckUpdate --> rowNum:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Likp:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectLipsCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Lips:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Lips:rowids"));
			Lips build = Lips.builder().rowids(rowids).build();

			List<Lips> slist = sapMapper.selectLipsCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Lips:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Lips target = gpMapper.selectLips(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getErdat() != null && source.getErzet() != null) {
							String erdat2 = new StringBuffer(source.getErdat()).insert(4, "-").insert(7, "-").toString();
							String erzet2 = new StringBuffer(source.getErzet()).insert(2, ":").insert(5, ":").toString();
							String dateStr = erdat2 + " " + erzet2;
							Date date = DateUtil.parse(dateStr);
							source.setErdat1(date);
							source.setErdat2(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateLips(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<Vbuk> sVbukList = sapMapper.cascadeVbukByLips(source); // 级联查询源库afko表
						for (Vbuk sVbuk : sVbukList) {
							Vbuk tVbuk = gpMapper.selectVbuk(sVbuk); // 查询目标库中afko表
							cascadeVbukCheckByUpdate(sVbuk, tVbuk); // 级联更新afko表
						}
						// 级联更新业务
						List<Vbup> sVbupList = sapMapper.cascadeVbupByLips(source); // 级联查询源库vbup表
						for (Vbup sVbup : sVbupList) {
							Vbup tVbup = gpMapper.selectVbup(sVbup); // 查询目标库中vbup表
							cascadeVbupCheckByUpdate(sVbup, tVbup); // 级联更新vbup表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Lips:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectLipscheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Lips:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectVbakCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Vbak:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Vbak:rowids"));
			Vbak build = Vbak.builder().rowids(rowids).build();

			List<Vbak> slist = sapMapper.selectVbakCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:Vbak:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				
				Map<String, Object> joinItem = sapMapper.selectVbak4Join(source);
				if (joinItem != null) { // join的查询替代方式
					Object object = joinItem.get("pernr");
					if (object != null) {
						source.setPernr(object.toString());
					}
				}
				
				Vbak target = gpMapper.selectVbak(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						source.setAudat1(SomeUtils.caDate(source.getAudat())); // 日期00000000格式转换,已处理异常
						// ===============================
						while (true) {
							try {
								String erdat = source.getErdat();
								String erzet = source.getErzet();
								if ("00000000".equals(erdat)) {
									source.setErdat2erzet(null);
								} else {
									Date date = DateUtil.parse(erdat+erzet, "yyyyMMddHHmmss");
									source.setErdat2erzet(date);
								}
								gpMapper.updateVbak(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<Konv> sKonvList = sapMapper.cascadeKonvByVbak(source); // 级联查询源库afko表
						for (Konv sKonv : sKonvList) {
							Konv tKonv = gpMapper.selectKonv(sKonv); // 查询目标库中afko表
							cascadeKonvCheckByUpdate(sKonv, tKonv); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Vbak:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectVbakcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Vbak:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectVbrkCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Vbrk:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Vbrk:rowids"));
			Vbrk build = Vbrk.builder().rowids(rowids).build();

			List<Vbrk> slist = sapMapper.selectVbrkCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Vbrk:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Vbrk target = gpMapper.selectVbrk(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(null);
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (source.getFkdat() != null) {
							String erdat2 = new StringBuffer(source.getFkdat()).insert(4, "-").insert(7, "-").toString();
							Date date = DateUtil.parse(erdat2);
							source.setFkdat1(date);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateVbrk(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<Vbrp> sVbrpList = sapMapper.cascadeVbrpByVbrk(source); // 级联查询源库afko表
						for (Vbrp sVbrp : sVbrpList) {
							Vbrp tVbrp = gpMapper.selectVbrp(sVbrp); // 查询目标库中afko表
							cascadeVbrpCheckByUpdate(sVbrp, tVbrp); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Vbrk:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectVbrkcheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Vbrk:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectSalesContractProcessCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:SalesContractProcess:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:SalesContractProcess:rowids");
			SalesContractProcess build = SalesContractProcess.builder().id(rowids).build();
			
			List<SalesContractProcess> slist = crmMapper.selectSalesContractProcessCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:SalesContractProcess:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				SalesContractProcess target = gpMapper.selectSalesContractProcess(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId();
				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.updateSalesContractProcess(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<SalesContractProcessMX> sSalesContractProcessMXList = crmMapper.cascadeSalesContractProcessMXBySalesContractProcess(source); // 级联查询源库afko表
						for (SalesContractProcessMX sSalesContractProcessMX : sSalesContractProcessMXList) {
							SalesContractProcessMX tSalesContractProcessMX = gpMapper.selectSalesContractProcessMX(sSalesContractProcessMX); // 查询目标库中afko表
							cascadeSalesContractProcessMXCheckByUpdate(sSalesContractProcessMX, tSalesContractProcessMX); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:SalesContractProcess:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectSalesContractProcesscheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:SalesContractProcess:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	public void selectSampleApplicationProcessCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:SampleApplicationProcess:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:SampleApplicationProcess:rowids");
			SampleApplicationProcess build = SampleApplicationProcess.builder().id(rowids).build();
			
			List<SampleApplicationProcess> slist = crmMapper.selectSampleApplicationProcessCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:SampleApplicationProcess:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				SampleApplicationProcess target = gpMapper.selectSampleApplicationProcess(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId();
				if (target != null) {
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						String pcbcs = source.getPCBCS();
						String chanPinYY = source.getChanPinYY();
						if (StrUtil.isBlank(pcbcs) || StrUtil.isBlank(chanPinYY)) {
							source.setPCBCS2chanPinYY(null);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateSampleApplicationProcess(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<SampleApplicationProcessMingXi> sSampleApplicationProcessMingXiList = crmMapper.cascadeSampleApplicationProcessMingXiBySampleApplicationProcess(source); // 级联查询源库afko表
						for (SampleApplicationProcessMingXi sSampleApplicationProcessMingXi : sSampleApplicationProcessMingXiList) {
							SampleApplicationProcessMingXi tSampleApplicationProcessMingXi = gpMapper.selectSampleApplicationProcessMingXi(sSampleApplicationProcessMingXi); // 查询目标库中afko表
							cascadeSampleApplicationProcessMingXiCheckByUpdate(sSampleApplicationProcessMingXi, tSampleApplicationProcessMingXi); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:SampleApplicationProcess:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectSampleApplicationProcesscheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:SampleApplicationProcess:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	public void selectSalesInquiryProcessCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:SalesInquiryProcess:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:SalesInquiryProcess:rowids");
			SalesInquiryProcess build = SalesInquiryProcess.builder().id(rowids).build();
			
			List<SalesInquiryProcess> slist = crmMapper.selectSalesInquiryProcessCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:SalesInquiryProcess:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				SalesInquiryProcess target = gpMapper.selectSalesInquiryProcess(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId();
				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.updateSalesInquiryProcess(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<SalesInquiryProcessMX> sSalesInquiryProcessMXList = crmMapper.cascadeSalesInquiryProcessMXBySalesInquiryProcess(source); // 级联查询源库
						for (SalesInquiryProcessMX sSalesInquiryProcessMX : sSalesInquiryProcessMXList) {
							SalesInquiryProcessMX tSalesInquiryProcessMX = gpMapper.selectSalesInquiryProcessMX(sSalesInquiryProcessMX); // 查询目标库中
							cascadeSalesInquiryProcessMXCheckByUpdate(sSalesInquiryProcessMX, tSalesInquiryProcessMX); // 级联更新
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:SalesInquiryProcess:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectSalesInquiryProcesscheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:SalesInquiryProcess:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	public void selectSampleClosingProcessCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:SampleClosingProcess:rowids", "0");
			String rowids = opsForValue.get("huazheng:checkUpdate:SampleClosingProcess:rowids");
			SampleClosingProcess build = SampleClosingProcess.builder().id(rowids).build();
			
			List<SampleClosingProcess> slist = crmMapper.selectSampleClosingProcessCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) {
				redis1Template.opsForValue().set("huazheng:checkUpdate:SampleClosingProcess:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}
			
			slist.forEach(source -> { // 遍历要检查的数据
				SampleClosingProcess target = gpMapper.selectSampleClosingProcess(source); // 根据主键查询源库中的数据
				String operator = "none";
				String srowids = source.getId();
				if (target != null) {
					String shash = SecureUtil.md5(JSONUtil.toJsonStr(source)); // 源库中数据的hash结果
					String thash = target.getHashResult(); // 数仓中数据的hash结果
					if (!shash.equals(thash)) { // 如果hash结果不一致
						source.setHashResult(shash);
						// ===============================
						if (NumberUtil.isNumber(source.getYangPinSQSLH())) {
							int yangPinSQSLHi = Integer.parseInt(source.getYangPinSQSLH());
							source.setYangPinSQSLHi(yangPinSQSLHi);
						}
						// ===============================
						while (true) {
							try {
								gpMapper.updateSampleClosingProcess(source); // 更新数据到数仓中
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());ThreadUtil.safeSleep(500);
							}
						}
						
						// 级联更新业务
						List<SampleClosingProcessMingXi> sSampleClosingProcessMingXiList = crmMapper.cascadeSampleClosingProcessMingXiBySampleClosingProcess(source); // 级联查询源库afko表
						for (SampleClosingProcessMingXi sSampleClosingProcessMingXi : sSampleClosingProcessMingXiList) {
							SampleClosingProcessMingXi tSampleClosingProcessMingXi = gpMapper.selectSampleClosingProcessMingXi(sSampleClosingProcessMingXi); // 查询目标库中afko表
							cascadeSampleClosingProcessMingXiCheckByUpdate(sSampleClosingProcessMingXi, tSampleClosingProcessMingXi); // 级联更新afko表
						}
						
						ThreadUtil.safeSleep(500);
					}
				}
				redis1Template.opsForValue().set("huazheng:checkUpdate:SampleClosingProcess:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectSampleClosingProcesscheckUpdate --> rowids:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:SampleClosingProcess:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
	
	public void selectTasksCheckUpdate() {
		try {
			ValueOperations<String, String> opsForValue = redis1Template.opsForValue();
			opsForValue.setIfAbsent("huazheng:checkUpdate:Tasks:rowids", "0");
			Long rowids = Long.parseLong(opsForValue.get("huazheng:checkUpdate:Tasks:rowids"));
			Tasks build = Tasks.builder().rowids(rowids).build();

			List<Tasks> slist = ultimusDBMapper.selectTasksCheckByUpdate(build); // 从数仓中查询一组数据
			if (slist.size() == 0) { // rowid已完成，则转为由更新时间字段开始同步
				redis1Template.opsForValue().set("huazheng:checkUpdate:Tasks:rowids", "0"); // 计数器复位
				ThreadUtil.sleep(1000); // 没有数据了，休眠一下
			}

			slist.forEach(source -> { // 遍历要检查的数据
				Tasks target = gpMapper.selectTasks(source); // 根据主键查询源库中的数据
				String operator = "none";
				Long srowids = source.getRowids();
				if (target != null) {
					source.setRowids(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.updateTasks(source); // 更新数据到数仓中
								operator = "update";
								break;
							} catch (RuntimeException e) {
								log.error(e.getMessage());
								ThreadUtil.safeSleep(500);
							}
						}
						ThreadUtil.safeSleep(500);
					}
				} else { // 不能自增，在更新的时候写入数据
					gpMapper.insertTasks(source);
				}

				redis1Template.opsForValue().set("huazheng:checkUpdate:Tasks:rowids", srowids.toString());
				if (!operator.equals("none")) {
					log.info(String.format("selectTaskscheckUpdate --> rowNum:%s, operator:%s", srowids.toString(), operator));
				}
			});
		} catch (Exception e) {
			redis1Template.opsForValue().set("huazheng:checkUpdateError:Tasks:rowids", SomeUtils.getErrorInfoFromException(e));
		}
	}
	
}
