pom.xml

@@ -34,6 +34,10 @@
     </dependencyManagement>
     <dependencies>
         <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-quartz</artifactId>
+        </dependency>
+        <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
         </dependency>

src/main/java/com/gct/tools/etlcamelhuge/MQ/DiagnoseMessageProducer.java

@@ -49,7 +49,7 @@ public class DiagnoseMessageProducer implements MessageProducer {
             return;
         }
         DiagnoseMsg actualMsg = (DiagnoseMsg) msgBody;
-        System.out.println(properties);
+//        System.out.println(properties);
         Message msg = new Message(properties.getTopic(), properties.getTags(), actualMsg.getWellId(), JSONObject.toJSONString(actualMsg).getBytes(StandardCharsets.UTF_8));
         sendDefault(producer, msg, (MessageBody) actualMsg, failCallBack, successCallBack,0);
     }

src/main/java/com/gct/tools/etlcamelhuge/camelconfig/QuartzConfig.java

@@ -0,0 +1,30 @@
+package com.gct.tools.etlcamelhuge.camelconfig;
+
+import com.gct.tools.etlcamelhuge.quartz.SyncWellSourceTask;
+import org.quartz.*;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static org.quartz.CronScheduleBuilder.cronSchedule;
+
+@Configuration
+public class QuartzConfig {
+
+
+    @Bean
+    public JobDetail jobDetail() {
+        return JobBuilder.newJob(SyncWellSourceTask.class)
+                .withIdentity("ModbusReadTask" ,"Group1")
+                .storeDurably()
+                .build();
+    }
+
+    @Bean
+    public Trigger trigger() {
+        return TriggerBuilder.newTrigger()
+                .forJob("ModbusReadTask", "Group1")
+                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
+                        .withIntervalInHours(1)  // 间隔1小时
+                        .repeatForever()).build();
+    }
+}

src/main/java/com/gct/tools/etlcamelhuge/quartz/SyncWellSourceTask.java

@@ -0,0 +1,106 @@
+package com.gct.tools.etlcamelhuge.quartz;
+
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.scheduling.quartz.QuartzJobBean;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.Resource;
+import javax.sql.DataSource;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class SyncWellSourceTask extends QuartzJobBean {
+
+    @Resource(name = "oracle")
+    DataSource oracle;
+
+    @Resource(name = "centralbase")
+    DataSource centralbase;
+
+    @Transactional(rollbackFor = Exception.class)
+    public void orgUpdate(){
+        JdbcTemplate oracleTemplate = new JdbcTemplate(oracle);
+        final JdbcTemplate centralbaseTemplate = new JdbcTemplate(centralbase);
+        String sql = String.format("select * from zy_dxgw.V_PC_ORGANIZATION_T where CANTON = '重油公司'");
+        List<Map<String, Object>> results = oracleTemplate.queryForList(sql);
+
+        Map<String, Map<String, Object>> parentObjMap = new HashMap<>();
+        Map<String, Map<String, Object>> orgIdMap = new HashMap<>();
+        AtomicInteger a = new AtomicInteger(1);
+        results.forEach(rowMap -> {
+            rowMap.put("org_id_map_int",a.getAndIncrement());
+            orgIdMap.put(rowMap.get("ORG_ID").toString(), rowMap);
+        });
+        results.forEach(rowMap -> {
+            parentObjMap.put(rowMap.get("ORG_ID").toString(), orgIdMap.get(rowMap.get("PARENT_ID")));
+        });
+
+        results.forEach(rowMap -> {
+            Stack<String> orgPreStack = new Stack<>();
+            Map<String, Object> parentObj = rowMap;
+
+            while ((parentObj = parentObjMap.get(parentObj.get("ORG_ID").toString())) != null) {
+                orgPreStack.push(parentObj.get("ORG_NAME").toString());
+            }
+            StringBuilder orgIdPre = new StringBuilder();
+            while (!orgPreStack.isEmpty()) {
+                orgIdPre.append(orgPreStack.pop()).append("@");
+            }
+            orgIdPre.append(rowMap.get("ORG_NAME"));
+            String orgIdPreStr = orgIdPre.toString();
+            rowMap.put("ORG_ID_PRE",orgIdPreStr);
+
+         /*   String sqlInsertOrg = String.format("insert into centralbase.cb_pc_organization(" +
+                            "org_id,org_name,org_code,org_level,parent_id,org_id_pre) " +
+                            "values('%s','%s','%s','%s','%s','%s')", rowMap.get("org_id_map_int"), rowMap.get("ORG_NAME"), rowMap.get("ORG_CODE"),
+                    orgIdPreStr.split("@").length
+                    , Objects.nonNull(orgIdMap.get(rowMap.get("PARENT_ID").toString()))?orgIdMap.get(rowMap.get("PARENT_ID").toString()).get("org_id_map_int"):"0"
+                    , orgIdPreStr);
+//            System.out.println(sqlInsertOrg);
+            centralbaseTemplate.execute(sqlInsertOrg);  */ //excute finish, unused
+        });
+
+        //excute wellsource
+        sql = String.format("select WELL_ID,well_common_name,org_id from " +
+                "zy_dxgw.pc_dev_well_attr_info where well_purpose_name LIKE '%%采油井%%'");
+        results = oracleTemplate.queryForList(sql);
+        results.forEach(rowMap -> {
+            String wellId = rowMap.get("WELL_ID").toString();
+            String wellCommonName = rowMap.get("WELL_COMMON_NAME").toString();
+            String wellLegalName = rowMap.get("WELL_COMMON_NAME").toString();
+            String orgId = rowMap.get("ORG_ID").toString();
+            String orgIdMapInt = orgIdMap.get(orgId).get("org_id_map_int").toString();
+            String stationId = orgIdMap.get(orgId).get("ORG_ID_PRE").toString();
+
+
+            String sqlQueryWellSource = String.format("select * from centralbase.cb_cd_well_source where well_id = '%s'",wellId);
+            List<Map<String, Object>> maps = centralbaseTemplate.queryForList(sqlQueryWellSource);
+            if(maps.isEmpty()){
+                System.out.println("===========> 添加新井: "+wellCommonName);
+                String  sqlInsertWellSource = String.format("insert into centralbase.cb_cd_well_source(well_id,well_common_name," +
+                                "well_legal_name,org_id,station_id) select '%s','%s','%s','%s','%s' " +
+                                "where not exists (select * from centralbase.cb_cd_well_source where well_id = '%s')",
+                        wellId, wellCommonName, wellLegalName, orgIdMapInt, stationId, wellId);
+//            System.out.println(sqlInsertWellSource);
+                centralbaseTemplate.execute(sqlInsertWellSource);
+
+                String sqlInsertSysWellControl = String.format("insert into centralbase.sys_access_well_control(well_id," +
+                        "well_common_name,org_id,access_date,access_status) select '%s','%s','%s','%s','%s' " +
+                                " where not exists (select * from centralbase.sys_access_well_control where well_id = '%s')",
+                        wellId, wellCommonName, orgIdMapInt,
+                        LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),"1",wellId);
+                centralbaseTemplate.execute(sqlInsertSysWellControl);
+            }
+        });
+    }
+
+    @Override
+    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
+        orgUpdate();
+    }
+}
+

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

@@ -67,7 +67,7 @@ public class CamelJDBCCofRealTimeConfiguration {
     public void sendDataToRocketMQ(String wellName, String wellId, String prodDate, Double stroke_length, Double stroke_frequency, String sgt) {
         String orgId = "0";
         DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), stroke_length, stroke_frequency);
-        System.out.println(diagnoseMsg);
+//        System.out.println(diagnoseMsg);
         sendMsgRunTime++;
         producer.send(diagnoseMsg);
     }

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

@@ -52,6 +52,7 @@ public class CamelJDBCConfiguration {
             @Override
             public void configure() throws Exception {
 
+
                 from("timer:update-wellControl?period=3600000")
                         .routeId("update-wellControl")
                         .setBody(simple("select scc.well_id,wo.well_common_name,op.org_id from centralbase.sys_access_well_control scc\n" +
@@ -168,7 +169,10 @@ public class CamelJDBCConfiguration {
                         .split(body())
                         .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set pump_depth = '${body[PUMP_DEPTH]}'  " +
                                 "where well_id = '${header.well_id}' and prod_date::date  = '${header.date}' "))
+                        .doTry()
                         .to("jdbc:centralbase")
+                        .doCatch(Exception.class)
+                        .log("${header.date}" + " routeId:update-statusDaily-BS ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
                         .end();
 
                 RouteDefinition statusDailyBJ = from("timer:mytimer-update-statusDaily-BJ?period=10800000")

src/main/resources/application.yml

@@ -158,4 +158,7 @@ gct:
 swagger:
   authorization:
     key-name: token
-  enabled: true
+  enabled: true
+logging:
+  level:
+    root: error