同步井信息

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/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,102 @@
+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 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;
+
+    private 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
+        });
+
+        if(a.get()>1)return;
+        //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='采油井'");
+        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) values('%s','%s','%s','%s','%s')",
+                        wellId, wellCommonName, wellLegalName, orgIdMapInt, stationId);
+//            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) values('%s','%s','%s','%s,'%s')",
+                        wellId, wellCommonName, orgIdMapInt, LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),"1");
+                centralbaseTemplate.execute(sqlInsertSysWellControl);
+            }
+        });
+    }
+
+    @Override
+    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
+        orgUpdate();
+    }
+}
+

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

@@ -51,113 +51,6 @@ public class CamelJDBCConfiguration {
             @Override
             public void configure() throws Exception {
 
-                /*RouteDefinition OrgAndWellSource= (RouteDefinition) from("timer:insert-OrgAndWellSource?period=86400000")
-                        .routeId("insert-OrgAndWellSource")
-                        .setHeader("date", constant(getDate()+" 00:00:00"))
-                        .process(exchange -> {
-                            org = 0;
-                            orgIdPreList = new HashMap<>();
-                        })
-                        .setBody(simple("select max(org_id) from centralbase.cb_pc_organization"))
-                        .to("jdbc:centralbase")
-                        .process(exchange -> {
-                            HashMap body = exchange.getIn().getBody(HashMap.class);
-                            if (body == null || StringUtils.isEmpty(body.get("max"))) org = 0;
-                            else
-                                org = Integer.valueOf(body.get("max").toString());
-                        })
-                        .setBody(simple("select org_id_pre,org_id from centralbase.cb_pc_organization"))
-                        .to("jdbc:centralbase")
-                        .split(body()).process(exchange -> {
-                            HashMap body = exchange.getIn().getBody(HashMap.class);
-                            orgIdPreList.put(body.get("org_id_pre"),body.get("org_id"));
-                        }).end();
-                        setMyBody(OrgAndWellSource)
-                       .setBody(simple("select  distinct WELL_ID,cydmc,zyq,zk,qyrq,sccw,qk,REMARKS from DBA01 where PROD_DATE=to_date('${header.date}','yyyy-mm-dd hh24:mi:ss') and WELL_ID ='${header.well_id}'"))
-                       .to("jdbc:oracle")
-                       .transform()
-                       .body((result) -> {
-                              organization = new TreeSet<>();
-                              return result;
-                          })
-                       .step("1")
-                       .split(body()).process(exchange -> {
-                      Message in = exchange.getIn();
-                      HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                      if (StringUtils.isEmpty(aRow.get("qyrq"))){
-                          aRow.put("QYRQ","2021-01-01 00:00:00");
-                      }
-                      if (!aRow.containsKey("JM")) aRow.put("JM",aRow.get("WELL_ID"));
-                      String org_level3 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC") + "@" + aRow.get("ZK");
-                      String org_level2 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC");
-                      String org_level1 = aRow.get("ZYQ").toString();
-                      aRow.put("station_id", org_level3);
-                      orgID = org;
-                      if ((!orgIdPreList.containsKey(org_level1)) || (!orgIdPreList.containsKey(org_level2)) || (!orgIdPreList.containsKey(org_level3))) {
-                          if (organization.add(org_level1)) {
-                                if (!orgIDs.containsKey(org_level1)) orgIDs.put(org_level1,++orgID);
-                          }
-                          if (organization.add(org_level2)) {
-                              if (!orgIDs.containsKey(org_level2)) orgIDs.put(org_level2,++orgID);
-                          }
-                          if (organization.add(org_level3)) {
-                              if (!orgIDs.containsKey(org_level3)) orgIDs.put(org_level3,++orgID);
-                          }
-                      }
-                      if(orgIdPreList.get(org_level3) !=null){
-                          aRow.put("org_id",orgIdPreList.get(org_level3));
-                          return;
-                      }
-                      if(orgIDs.get(org_level3) !=null){
-                          aRow.put("org_id",orgIDs.get(org_level3));
-                      }
-                  })
-                  .setBody(simple("insert into centralbase.cb_cd_well_source (well_id,well_common_name,spud_date,org_id,station_id,station_name,completion_name,PRODUCING_AREA_name,remarks) " +
-                                  "select '${body[WELL_ID]}','${body[JM]}','${body[QYRQ]}'::timestamp,'${body[org_id]}','${body[station_id]}','${body[ZK]}','${body[SCCW]}','${body[QK]}','${body[REMARKS]}' " +
-                                  "where NOT EXISTS ( SELECT * FROM centralbase.cb_cd_well_source WHERE well_id = '${body[WELL_ID]}' )"))
-                  .to("jdbc:centralbase")
-                  .end()
-                  .transform().body((re) -> {
-                      List<Map<String, Object>> rows = new ArrayList<>();
-                      for (String s : organization) {
-                          Map<String, Object> row = new HashMap<>();
-                          String[] orgs = s.split("@");
-                          row.put("org_id_pre", s);
-                          switch (orgs.length) {
-                              case 1:
-                                  row.put("org_name", orgs[0]);
-                                  row.put("org_level", 1);
-                                  row.put("org_parent", "0");
-                                  break;
-                              case 2:
-                                  row.put("org_name", orgs[1]);
-                                  row.put("org_level", 2);
-                                  row.put("org_parent", orgIDs.get(orgs[0]).toString());
-                                  break;
-                              case 3:
-                                  row.put("org_name", orgs[2]);
-                                  row.put("org_level", 3);
-                                  row.put("org_parent", orgIDs.get(orgs[0] + "@" + orgs[1]).toString());
-                                  break;
-                          }
-                          if (!orgIdPreList.containsKey(s)) {
-                              org++;
-                              row.put("org_code", org);
-                              row.put("org_id", "" + org);
-                              orgIdPreList.put(s, row.get("org_id"));
-                              rows.add(row);
-                          }
-                      }
-                      return rows;
-                  }).split(body())
-                  .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_code,org_name,org_level,parent_id,org_id_pre)" +
-                                  "select '${body[org_id]}','${body[org_code]}','${body[org_name]}','${body[org_level]}','${body[org_parent]}','${body[org_id_pre]}' " +
-                                  "where NOT EXISTS ( SELECT * FROM centralbase.cb_pc_organization WHERE org_id = '${body[org_id]}')"))
-                  .doTry()
-                  .to("jdbc:centralbase")
-                  .doCatch(Exception.class)
-                  .log("${header.date}"+" routeId:insert-OrgAndWellSource->  centralbase.cb_pc_organization insert data failed")
-                  .end();*/
 
                 from("timer:update-wellControl?period=3600000")
                         .routeId("update-wellControl")

src/main/resources/application.yml

@@ -7,7 +7,7 @@ spring:
     ds1:
       ## Hikari连接池的设置 Hikari 时间单位都是毫秒
       type: com.zaxxer.hikari.HikariDataSource
-      jdbc-url: jdbc:postgresql://11.72.150.241:54321/database
+      jdbc-url: jdbc:postgresql://10.72.118.135:54321/database
       username: root
       password: 123456
       driver-class-name: org.postgresql.Driver
@@ -30,7 +30,7 @@ spring:
     ds2:
       ## Hikari连接池的设置 Hikari 时间单位都是毫秒
       type: com.zaxxer.hikari.HikariDataSource
-      jdbc-url: jdbc:postgresql://11.72.150.241:54321/diagnosis
+      jdbc-url: jdbc:postgresql://10.72.118.135:54321/diagnosis
       username: root
       password: 123456
       driver-class-name: org.postgresql.Driver
@@ -75,7 +75,7 @@ spring:
     ds4:
       ## Hikari连接池的设置 Hikari 时间单位都是毫秒
       type: com.zaxxer.hikari.HikariDataSource
-      jdbc-url: jdbc:postgresql://11.72.150.241:54321/calliquid
+      jdbc-url: jdbc:postgresql://10.72.118.135:54321/calliquid
       username: root
       password: 123456
       driver-class-name: org.postgresql.Driver
@@ -118,7 +118,7 @@ server:
 
 rocketmq:
   #rocketmq的路由调度器的地址
-  name-server: 11.72.150.241:9876
+  name-server: 10.72.118.135:9876
   producer:
     # 消息分组
     group: aoid
@@ -136,7 +136,7 @@ gct:
         group: diagnose
         access-key: diagnose-msg-v1
         secret-key: diagnose-msg-v1
-      name-server: 11.72.150.241:9876
+      name-server: 10.72.118.135:9876
       consumer:
         group: diagnose
         access-key: diagnose-msg-v1
@@ -149,7 +149,7 @@ gct:
         group: warn
         access-key: warn-msg-v1
         secret-key: warn-msg-v1
-      name-server: 11.72.150.241:9876
+      name-server: 10.72.118.135:9876
       consumer:
         group: warn
         access-key: warn-msg-v1