bj,dym 重复导入不是最新问题解决

src/main/java/com/gct/tools/etlcamelhuge/routeconfig/CamelJDBCConfiguration.java

@@ -178,24 +178,26 @@ public class CamelJDBCConfiguration {
                         .setBody(simple("SELECT  distinct jh,max(rq),dym FROM DBA01 WHERE dym is not null group by jh,dym"))
                         .to("jdbc:oracle")
                         .split(body())
-                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set start_pump_liq_level = '${body[DYM]}' where well_id = '${body[JH]}' and prod_date::date  = '${header.date}' "))
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set start_pump_liq_level = '${body[DYM]}',dym_date ='${body[RQ]}'  where well_id = '${body[JH]}' and prod_date::date  = '${header.date}' and dym_date < '${body[RQ]}' "))
                         .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
                         .log("${header.date}"+" routeId:update-statusDaily-DYM ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
                         .end();
-
+                //因为数据库有可能会查出多条数据,现在的做法时,在数据库中增加两个个字段,用来记录BJ和DYM 的更新时间 ,只有获取到的时间大于数据库中存储的时间时,才会更新,并且该字段也会更新
                 from("timer:mytimer-update-statusDaily-BJ?period=1800000")
-                        .routeId("update-statusDaily-DYM")
+                        .routeId("update-statusDaily-BJ")
                         .setHeader("date", constant(getDate() + " 00:00:00"))
-                        .setBody(simple("SELECT  distinct jh,max(rq),bj FROM DBA01 WHERE dym is not null group by jh,dym"))
+                        .setBody(simple("SELECT  distinct jh,max(rq),bj FROM DBA01 WHERE bj is not null group by jh,bj"))
                         .to("jdbc:oracle")
-                        .split(body())
-                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '${body[BJ]}' where well_id = '${body[JH]}' and prod_date::date  = '${header.date}' "))
+                        .split(body()).process(exchange -> {
+                            HashMap body = exchange.getIn().getBody(HashMap.class);
+                        })
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '${body[BJ]}',bj_date = '${body[RQ]}' where well_id = '${body[JH]}' and prod_date::date  = '${header.date}' and bj_date < '${body[RQ]}'  "))
                         .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:update-statusDaily-DYM ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
+                        .log("${header.date}"+" routeId:update-statusDaily-BJ ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
                         .end();
 
                 from("timer:mytimer-update-statusDaily-submergenceDepth?period=1800000")