增加statusDaily想关接口

src/main/java/com/gct/tools/etlcamelhuge/EtlCamelHugeApplication.java

@@ -3,10 +3,12 @@ package com.gct.tools.etlcamelhuge;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableAsync;
 import springfox.documentation.swagger2.annotations.EnableSwagger2;
 
 @SpringBootApplication
 @EnableSwagger2
+@EnableAsync
 public class EtlCamelHugeApplication {
     public static void main(String[] args) {
         SpringApplication.run(EtlCamelHugeApplication.class, args);

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

@@ -0,0 +1,50 @@
+package com.gct.tools.etlcamelhuge.camelconfig;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.lang.reflect.Method;
+
+/**
+ * class name: AsyncConfiguration
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/25 下午1:45
+ */
+@Configuration
+@EnableAsync
+public class AsyncConfiguration implements AsyncConfigurer {
+
+    @Bean
+    public AsyncTaskExecutor asyncTaskExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(8);
+        executor.setMaxPoolSize(8);
+        return executor;
+    }
+    @Override
+    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler(){
+        return new LogAsyncUncaughtExceptionHandler();
+    }
+    class LogAsyncUncaughtExceptionHandler implements  AsyncUncaughtExceptionHandler{
+
+        @Override
+        public void handleUncaughtException(Throwable ex, Method method, Object... params) {
+            String paramLog="";
+            int i=0;
+            for (Object param : params) {
+                paramLog+="param:"+i+"=>"+param.toString()+" ";
+            }
+            Logger log = LoggerFactory.getLogger(this.getClass());
+            log.error("Method:{} run error,Exception Message:{},StackTrace:{},input params:{}",method.getName(),ex.getMessage(),ex.getStackTrace(),paramLog);
+        }
+    }
+}

src/main/java/com/gct/tools/etlcamelhuge/controller/BaseDataController.java

@@ -0,0 +1,118 @@
+package com.gct.tools.etlcamelhuge.controller;
+
+import com.alibaba.fastjson.JSONObject;
+import com.gct.tools.etlcamelhuge.entity.BaseDataBody;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.omg.CORBA.IRObject;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * class name: BaseDataController
+ *
+ * @author gxt
+ * @version 1.0
+ * @since 2021/8/30 下午3:51 周一
+ */
+@RestController
+@RequestMapping("/BaseData")
+@Api(value = "BaseDataController",description = "基础数据操作controller")
+public class BaseDataController {
+
+    private JdbcTemplate jdbcTemplate;
+    @Resource(name = "oracle")
+    DataSource oracleDataSource;
+
+    @Resource(name = "centralbase")
+    DataSource baseDataSource;
+
+    @PostMapping("/StatusDaily")
+    @ApiOperation("从 A2 获取 数据插入到 centrlBase 中")
+    public JSONObject saveDataToStatusDaily(@RequestBody BaseDataBody baseDataBody){
+        JSONObject jsonObject = new JSONObject();
+        long insertCount = 0;
+        try {
+            jdbcTemplate = new JdbcTemplate(oracleDataSource);
+            String date = baseDataBody.getDate();
+            String sql = "select  distinct jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from DBA01 where rq  = to_date('"+date+"','yyyy-MM-dd') and qyrq is not null ";
+            List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
+            for (Map<String, Object> map : list) {
+                if (map.get("YZ") == null) map.put("YZ", "0.0");
+                if (map.get("HYSX") == null) map.put("HYSX", "0.0");
+                if (map.get("YYSX") == null) map.put("YYSX", "0.0");
+                if (map.get("TYSX") == null) map.put("TYSX", "0.0");
+                if (map.get("BS") == null) map.put("BS", "0.0");
+                sql = "insert into centralbase.cb_pc_pro_wellbore_status_daily(well_id,prod_date,oil_prod_method,oil_nozzle,back_pres,tubing_pres,casing_pres,pump_depth) " +
+                        "value (?,?,?,?,?,?,?,?)";
+                int count = insertBaseDataSourceOfStatusDaily(sql, map);
+                List<Map<String, Object>> dymIsNotNUllList = selectA2DataOfNotIsNullDYM();
+                for (Map<String, Object> stringObjectMap : dymIsNotNUllList) {
+                    sql = "update centralbase.cb_pc_pro_wellbore_status_daily set start_pump_liq_level = '"+stringObjectMap.get("DYM")+"' where well_id = '"+stringObjectMap.get("JH")+"' and prod_date::date  = '"+stringObjectMap.get("RQ")+"' ";
+                    int dymIsNotNullCount = updateBaseDataSourceOfStatusDaily(sql);
+                    insertCount = insertCount + dymIsNotNullCount;
+                }
+                sql = "select well_id,prod_date,start_pump_liq_level,pump_depth from centralbase.cb_pc_pro_wellbore_status_daily where prod_date = '"+map.get("RQ")+"' ";
+                List<Map<String, Object>> CMDDataList = selectBaseDataSourceStatusDaily(sql);
+                for (Map<String, Object> stringObjectMap : CMDDataList) {
+                    stringObjectMap.put("submergence_depth",null);
+                    if (stringObjectMap.get("start_pump_liq_level")!=null && stringObjectMap.get("pump_depth")!=null){
+                        double cmd= Double.valueOf(stringObjectMap.get("pump_depth").toString())-Double.valueOf(stringObjectMap.get("start_pump_liq_level").toString())/10;
+                        BigDecimal bd=new BigDecimal(cmd);
+                        double cmd1=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                        stringObjectMap.put("submergence_depth",cmd1);
+                    }
+                    sql = " update centralbase.cb_pc_pro_wellbore_status_daily set submergence_depth = '"+stringObjectMap.get("submergence_depth")+"' where well_id = '"+stringObjectMap.get("well_id")+"' and prod_date  = '"+stringObjectMap.get("prod_date")+"' ";
+                    int CMDUpdateCount = updateBaseDataSourceOfStatusDaily(sql);
+                    insertCount = insertCount + CMDUpdateCount;
+                }
+                sql = " select distinct rn.well_id,cb.prod_date,rn.pump_diameter  from centralbase.cb_temp_well_mech_runtime rn ,centralbase.cb_pc_pro_wellbore_status_daily cb where cb.well_id = rn.well_id " +
+                        " and cb.prod_date = '"+date+"'  ";
+                List<Map<String, Object>> oilNozzleList = selectBaseDataSourceStatusDaily(sql);
+                for (Map<String, Object> stringObjectMap : oilNozzleList) {
+                    sql = " update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '"+stringObjectMap.get("pump_diameter")+"' where well_id ='"+stringObjectMap.get("well_id")+"' and prod_date='"+stringObjectMap.get("prod_date")+"' ";
+                    int oliNozzle = updateBaseDataSourceOfStatusDaily(sql);
+                    insertCount = insertCount + oliNozzle;
+                }
+            }
+        }catch (Exception e){
+            e.printStackTrace();
+            jsonObject.put("error",e.getMessage());
+        }finally {
+            jsonObject.put("插入条数",insertCount);
+        }
+        return jsonObject;
+    }
+
+
+    public int insertBaseDataSourceOfStatusDaily(String sql,Map<String,Object> map){
+        jdbcTemplate = new JdbcTemplate(baseDataSource);
+        if (map.isEmpty()) return 0;
+        return  jdbcTemplate.update(sql, map.get("JH"), map.get("RQ"), map.get("CYFS"), map.get("YZ"), map.get("HYSX"), map.get("YYSX"), map.get("BS"), map.get("DYM"));
+    }
+
+    public List<Map<String, Object>> selectA2DataOfNotIsNullDYM(){
+        jdbcTemplate = new JdbcTemplate(oracleDataSource);
+        String sql = "SELECT  distinct jh,max(rq),dym FROM DBA01 WHERE dym is not null group by jh,dym";
+        return  jdbcTemplate.queryForList(sql);
+    }
+
+    public List<Map<String, Object>> selectBaseDataSourceStatusDaily(String sql){
+        jdbcTemplate = new JdbcTemplate(baseDataSource);
+         return  jdbcTemplate.queryForList(sql);
+    }
+    public int updateBaseDataSourceOfStatusDaily(String sql){
+        jdbcTemplate = new JdbcTemplate(baseDataSource);
+        return  jdbcTemplate.update(sql);
+    }
+
+}

src/main/java/com/gct/tools/etlcamelhuge/controller/GtController.java

@@ -30,7 +30,7 @@ import java.util.stream.Collectors;
  * @since 2021/8/26 下午2:50 周四
  */
 @RestController
-@RequestMapping("/")
+@RequestMapping("/GTData")
 @Api(value = "GtController",description = "功图数据操作controller")
 public class GtController {
 
@@ -38,6 +38,9 @@ public class GtController {
     @Resource(name = "centralbase")
     DataSource dataSource;
 
+    @Resource(name = "gtsj")
+    DataSource dataSourceOfGTSJ;
+
     @Resource(name = "diagnoseMessageProducer")
     private MessageProducer producer;
 
@@ -49,7 +52,7 @@ public class GtController {
         List<String> wellList = gtBody.getWellList().stream().map(x->x.toString()).collect(Collectors.toList());
         JSONObject jsonObject = new JSONObject();
         jdbcTemplate = new JdbcTemplate(dataSource);
-        int sumData = 0;
+        long sumData = 0;
         try {
                 if (wellList.isEmpty() || wellList.size()==0 ){
                     String sql = String.format("select so.well_id,so.well_common_name,so.org_id,ti.prod_date,ti.stroke_frequency,ti.stroke_length,ti.sgt from centralbase.cb_temp_well_mech_runtime ti, centralbase.cb_cd_well_source so " +
@@ -71,7 +74,7 @@ public class GtController {
         return  jsonObject;
     }
 
-    public int sendDataToMQ(String sql){
+    public long sendDataToMQ(String sql){
         int sumData = 0;
         List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
         for (Map<String, Object> map : list) {
@@ -92,15 +95,15 @@ public class GtController {
         return sumData;
     }
 
-   /* @GetMapping("/getGTSJ")
+    @GetMapping("/getGTSJ")
     @ApiOperation(value = "从实时的机采数据生产的表中获取数据放入到 Runtime 表中")
     public JSONObject getGTSJ(@RequestBody GTBody gtBody){
 
         String startDate = gtBody.getStartDate();
         String endDate = gtBody.getEndDate();
-        List<String> wellList = gtBody.getWellList();
+        JSONArray wellList = gtBody.getWellList();
         JSONObject jsonObject = new JSONObject();
-        jdbcTemplate = new JdbcTemplate(dataSource);
+        jdbcTemplate = new JdbcTemplate(dataSourceOfGTSJ);
         int curPage = 0;
         int pageSize = 5000;
         int sumData = 0;
@@ -117,13 +120,13 @@ public class GtController {
                         Double susp_max_load = max(disp_loads);
                         Double susp_min_load = min(disp_loads);
                         String sgt = "";
-                        for (int i = 0; i < displacements.length; i++) {
-                            sgt = sgt + displacements[i] + "," + disp_loads[i] + ",";
+                        for (int y = 0; y < displacements.length; y++) {
+                            sgt = sgt + displacements[y] + "," + disp_loads[y] + ",";
                         }
                         String[] s = sgt.split(",");
                         String w = "";
-                        for (int i = 0; i < s.length; i++) {
-                            w += new BigDecimal(Math.round(Double.parseDouble(s[i]) * 100)).stripTrailingZeros().toPlainString() + ",";
+                        for (int x = 0; x < s.length; x++) {
+                            w += new BigDecimal(Math.round(Double.parseDouble(s[x]) * 100)).stripTrailingZeros().toPlainString() + ",";
                         }
                         Double[][] doubles = SGTUtil.encodeToDoubleArray(w);
                         map.put("sgt", SGTUtil.encodeToString(doubles));
@@ -148,11 +151,11 @@ public class GtController {
                     jdbcTemplate.update("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,stroke_length,stroke_frequency,susp_max_load,susp_min_load,sgt) " +
                             "value (?,?,?,?,?,?,?)",map.get("well_name").toString(),map.get("dyna_create_time"),map.get("stroke"),map.get("frequency"),map.get("susp_max_load"),map.get("susp_min_load"),map.get("sgt"));
                 }
-               *//* sumData += list.size();
+                sumData += list.size();
                 if (list.size()< pageSize){
                     break;
                 }
-                curPage ++ ;*//*
+                curPage ++ ;
             }
         }catch (Exception e){
             jsonObject.put("error",e.getMessage());
@@ -160,7 +163,7 @@ public class GtController {
             jsonObject.put("sumData",sumData);
         }
         return    jsonObject;
-         }*/
+         }
 
 
     public Double min(String[] strings){

src/main/java/com/gct/tools/etlcamelhuge/entity/BaseDataBody.java

@@ -0,0 +1,25 @@
+package com.gct.tools.etlcamelhuge.entity;
+
+import com.alibaba.fastjson.JSONArray;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+/**
+ * class name: GTBody
+ *
+ * @author gxt
+ * @version 1.0
+ * @since 2021/8/27 下午2:34 周五
+ */
+@Data
+@ApiModel(value = "BaseDataBody", description = "")
+public class BaseDataBody {
+
+    @ApiModelProperty(value = "时间")
+    public String date;
+    /*@ApiModelProperty(value = "结束时间")
+    public String endDate;*/
+    @ApiModelProperty(value = "井号集合")
+    public JSONArray wellList;
+}

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

@@ -0,0 +1,220 @@
+package com.gct.tools.etlcamelhuge.routeconfig;
+
+import com.gct.common.util.SGTUtil;
+import com.gct.tools.etlcamelhuge.MQ.MessageProducer;
+import com.gct.tools.etlcamelhuge.entity.DiagnoseMsg;
+import lombok.Data;
+import org.apache.camel.Message;
+import org.apache.camel.builder.RouteBuilder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.Async;
+
+import javax.annotation.Resource;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+
+/**
+ * class name: CamelJDBCCofRealTimeConfiguration.java
+ *  实时导数据类
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/4/14 下午3:16
+ */
+@Configuration
+public class CamelJDBCCofRealTimeConfiguration  {
+
+    private static long sendMsgRunTime=0;
+
+    public Double min(String[] strings){
+        double[] doubles = new double[strings.length];
+        for (int i = 0; i < strings.length; i++) {
+            doubles[i] = Double.parseDouble(strings[i]);
+        }
+        return Arrays.stream(doubles).min().getAsDouble();
+    }
+    //获取最大载荷
+    public Double max(String[] strings){
+        double[] doubles = new double[strings.length];
+        for (int i = 0; i < strings.length; i++) {
+            doubles[i] = Double.parseDouble(strings[i]);
+        }
+        return Arrays.stream(doubles).max().getAsDouble();
+    }
+
+
+    public String getDate(){
+        return  LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
+    }
+
+
+    /***
+     * 异步发送 RunTime 消息到MQ中
+     * */
+    @Async
+    public long sendDataToRocketMQ(String wellName,String wellId,String prodDate,String stroke_length ,String stroke_frequency,String sgt){
+        String orgId = "0";
+        prodDate = prodDate.substring(0,19);
+        Double strokeLength = Double.valueOf(stroke_length);
+        Double strokeFrequency = Double.valueOf(stroke_frequency);
+        if (sgt == null || sgt.length() ==0){
+            sgt = "0,0";
+        }
+        DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), strokeLength, strokeFrequency);
+        sendMsgRunTime++;
+        producer.send(diagnoseMsg);
+        return sendMsgRunTime;
+    }
+
+    @Resource(name = "diagnoseMessageProducer")
+    private MessageProducer producer;
+    @Bean
+    public RouteBuilder routeBuilderWithRealTime() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() throws Exception {
+                from("timer:mytimer-insert-mechDaily?period=3600000")
+                        .routeId("insert-mech_daily")
+                        .setHeader("date", constant(getDate()))
+                        .setBody(simple("select distinct jh,rq,dym,jy,ly,bj,bs,bx,zs,cc,cs,blx,dl from DBA01 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null  "))
+                        .to("jdbc:oracle")
+                        .split(body()).process(exchange -> {
+                    Message in = exchange.getIn();
+                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                    if (aRow.get("JY") == null) aRow.put("JY", "0.0");
+                    if (aRow.get("LY") == null) aRow.put("LY", "0.0");
+                    if (aRow.get("BJ") == null) aRow.put("BJ", "0.0");
+                    if (aRow.get("BS") == null) aRow.put("BS", "0.0");
+                    if (aRow.get("BX") == null) aRow.put("BX", "0.0");
+                    if (aRow.get("ZS") == null) aRow.put("ZS", "0.0");
+                    if (aRow.get("CC") == null) aRow.put("CC", "0.0");
+                    if (aRow.get("CS") == null) aRow.put("CS", "0.0");
+                    if (aRow.get("BLX") == null) aRow.put("BLX", "");
+                    if (aRow.get("DL") == null) aRow.put("DL", "0.0");
+                })
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_daily(well_id,prod_date,static_pressure,flow_pres,pump_diameter,pump_depth,pump_efficiency,rotate_frequency,stroke_length,stroke_frequency,pump_type,elec_frequency) " +
+                                "select '${body[JH]}','${body[RQ]}','${body[JY]}','${body[LY]}','${body[BJ]}','${body[BS]}','${body[BX]}','${body[ZS]}','${body[CC]}','${body[CS]}','${body[BLX]}','${body[DL]}' " +
+                                "where NOT EXISTS ( SELECT * FROM centralbase.cb_temp_well_mech_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}' )"))
+                        .doTry()
+                        .to("jdbc:centralbase")
+                        .doCatch(Exception.class)
+                        .log("${header.date}"+" routeId:insert-mech_daily ->  centralbase.cb_temp_well_mech_daily insert data failed")
+                        .end();
+
+                from("timer:mytimer-update-runTime?period=3600000")
+                        .routeId("update-runTime")
+                        .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,bj"))
+                        .to("jdbc:oracle")
+                        .split(body()).process(exchange -> {
+                            Message in = exchange.getIn();
+                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                            if (aRow.get("BJ") ==null || aRow.get("BJ").equals("")) aRow.put("BJ","0.0");
+                        })
+                        .setBody(simple("update centralbase.cb_temp_well_mech_runtime set pump_diameter = '${body[BJ]}' where well_id = '${body[JH]}' "))
+                        .doTry()
+                        .to("jdbc:centralbase")
+                        .doCatch(Exception.class)
+                        .log("${header.date}"+" routeId:update-runTime -> centralbase.cb_temp_well_mech_runtime update data failed")
+                        .end();
+
+                from("timer:mytimer-insert-runtimeAndSendToMQ?period=3600000")
+                        .routeId("insert-runtimeAndSendToMQ")
+                        .setBody(simple("select max(prod_date) from centralbase.cb_temp_well_mech_runtime "))
+                        .to("jdbc:centralbase")
+                        .split(body())
+                        .setHeader("date", simple("${body[max]}"))
+                        .setBody(simple("select distinct well_name,dyna_create_time,check_date,displacement,disp_load,stroke,frequency,susp_max_load,susp_min_load from public.pc_fd_pumpjack_dyna_dia_t where   dyna_create_time > '${header.date}' "))
+                        .doTry()
+                        .to("jdbc:gtsj")
+                        .doCatch(Exception.class)
+                        .log("${deader.date}" + " routeId:insert-runtimeAndSendToMQ -> select runTime data failed")
+                        .process(exchange -> {
+                            sendMsgRunTime = 0;
+                        })
+                        .split(body()).process(exchange -> {
+                    Message in = exchange.getIn();
+                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                    String prod_date = aRow.get("dyna_create_time").toString().split("\\+")[0];
+                    aRow.put("dyna_create_time", prod_date);
+                    if (aRow.get("displacement") != null && !aRow.get("displacement").equals("") && aRow.get("disp_load") != null && !aRow.get("disp_load").equals("")) {
+                        String[] displacements = aRow.get("displacement").toString().split(";");//10 四舍五入
+                        String[] disp_loads = aRow.get("disp_load").toString().split(";");
+                        Double susp_max_load = max(disp_loads);
+                        Double susp_min_load = min(disp_loads);
+                        String sgt = "";
+                        for (int i = 0; i < displacements.length; i++) {
+                            sgt = sgt + displacements[i] + "," + disp_loads[i] + ",";
+                        }
+                        String[] s = sgt.split(",");
+                        String w = "";
+                        for (int i = 0; i < s.length; i++) {
+                            w += new BigDecimal(Math.round(Double.parseDouble(s[i]) * 100)).stripTrailingZeros().toPlainString() + ",";
+                        }
+                        Double[][] doubles = SGTUtil.encodeToDoubleArray(w);
+                        aRow.put("sgt", SGTUtil.encodeToString(doubles));
+                        aRow.put("susp_max_load",susp_max_load);
+                        aRow.put("susp_min_load",susp_min_load);
+                    }
+                    if (aRow.get("stroke") == null) aRow.put("stroke", "0.0");
+                    if (aRow.get("frequency") == null) aRow.put("frequency", "0.0");
+                    if (aRow.get("susp_max_load") == null) aRow.put("susp_max_load", "0.0");
+                    if (aRow.get("susp_min_load") == null) aRow.put("susp_min_load", "0.0");
+                    if (aRow.get("frequency") != null){
+                        BigDecimal bd=new BigDecimal(aRow.get("frequency").toString());
+                        double frequency=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                        aRow.put("frequency",frequency);
+                    }
+                    if (aRow.get("stroke") != null){
+                        double stroke1 = Double.parseDouble(aRow.get("stroke").toString());
+                        BigDecimal bd=new BigDecimal(stroke1);
+                        double stroke=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                        aRow.put("stroke",stroke);
+                    }
+                            long count = sendDataToRocketMQ(aRow.get("well_name").toString(), aRow.get("well_name").toString(), aRow.get("dyna_create_time").toString(), aRow.get("stroke").toString(), aRow.get("frequency").toString(), aRow.get("sgt").toString());
+                            System.out.println("消息发送时间为:   "+LocalDateTime.now()+"   发送的消息数量为: "+count + "   数据时间为:" + aRow.get("dyna_create_time"));
+
+                        })
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,stroke_length,stroke_frequency,susp_max_load,susp_min_load,sgt) " +
+                                "select '${body[well_name]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}' " +
+                                "where NOT EXISTS (SELECT * FROM centralbase.cb_temp_well_mech_runtime WHERE well_id = '${body[well_name]}' and  prod_date = '${body[dyna_create_time]}' )"))
+                        .doTry()
+                        .to("jdbc:centralbase")
+                        .doCatch(Exception.class)
+                        .log("${header.date}"+" routeId:insert-runtimeAndSendToMQ ->  centralbase.cb_temp_well_mech_runtime insert data failed ${body}")
+                        .end();
+
+               from("timer:mytimer-update-avg-mech_daily?period=3600000")
+                        .routeId("update-avg-mech_daily")
+                        .setHeader("date", constant(getDate()))
+                        .setBody(simple("select well_id,avg(stroke_length) stroke_length  ,avg(stroke_frequency) stroke_frequency from centralbase.cb_temp_well_mech_runtime where prod_date::date='${header.date}' group by well_id"))
+                        .to("jdbc:centralbase")
+                        .split(body()).process(exchange -> {
+                    Message in = exchange.getIn();
+                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                    if (aRow.get("stroke_length")!=null && aRow.get("stroke_frequency")!=null){
+                        double stroke_length=Double.parseDouble(aRow.get("stroke_length").toString());
+                        double stroke_frequency=Double.parseDouble(aRow.get("stroke_frequency").toString());
+                        BigDecimal bd=new BigDecimal(stroke_length);
+                        double stroke_lengt1=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                        BigDecimal bd1=new BigDecimal(stroke_frequency);
+                        double stroke_frequency1=bd1.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                        aRow.put("strokeLength",stroke_lengt1);
+                        aRow.put("strokeFrequency",stroke_frequency1);
+                    }
+                })
+                        .setBody(simple("update centralbase.cb_temp_well_mech_daily set stroke_length='${body[strokeLength]}' ,stroke_frequency ='${body[strokeFrequency]}' where well_id = '${body[well_id]}' and prod_date::date='${header.date}' "))
+                        .doTry()
+                        .to("jdbc:centralbase")
+                        .doCatch(Exception.class)
+                        .log("${header.date}"+" routeId:update-avg-mech_daily ->  centralbase.cb_temp_well_mech_daily update data failed")
+                        .end();
+            };
+        };
+    }
+}

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

@@ -16,6 +16,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.scheduling.annotation.Async;
 
 import javax.annotation.Resource;
 import javax.sql.DataSource;
@@ -41,37 +42,13 @@ class LogMessage{
  * @since 2021/4/14 下午3:16
  */
 @Configuration
-public class CamelJDBCConfiguration /*extends RouteBuilder */ {
-
-    public Double min(String[] strings){
-        double[] doubles = new double[strings.length];
-        for (int i = 0; i < strings.length; i++) {
-            doubles[i] = Double.parseDouble(strings[i]);
-        }
-        return Arrays.stream(doubles).min().getAsDouble();
-    }
-    //获取最大载荷
-    public Double max(String[] strings){
-        double[] doubles = new double[strings.length];
-        for (int i = 0; i < strings.length; i++) {
-            doubles[i] = Double.parseDouble(strings[i]);
-        }
-        return Arrays.stream(doubles).max().getAsDouble();
-    }
+public class CamelJDBCConfiguration {
 
 
     public String getDate(){
         return  LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
     }
 
-
-
-    private static Integer sendMsgRunTime=0;
-
-
-
-    @Resource(name = "diagnoseMessageProducer")
-    private MessageProducer producer;
     @Bean
     public RouteBuilder routeBuilderWithOracle1() {
         return new RouteBuilder() {
@@ -171,9 +148,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         .log("${header.date}"+" routeId:oracle-1->  centralbase.cb_pc_organization update data failed")
                         .end();*/
 
-
-                from("timer:mytimer2?period=3600000")
-                        .routeId("oracle-2")
+                from("timer:mytimer-insert-statusDaily?period=3600000")
+                        .routeId("insert-statusDaily")
                         .setHeader("date", constant(getDate()))
                         .setBody(simple("select  distinct jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from DBA01 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
                         .to("jdbc:oracle")
@@ -186,34 +162,32 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                     if (aRow.get("TYSX") == null) aRow.put("TYSX", "0.0");
                     if (aRow.get("BS") == null) aRow.put("BS", "0.0");
                 })
-                        .doTry()
                         .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_status_daily(well_id,prod_date,oil_prod_method,oil_nozzle,back_pres,tubing_pres,casing_pres,pump_depth) " +
                                 "select '${body[JH]}','${body[RQ]}','${body[CYFS]}','${body[YZ]}','${body[HYSX]}','${body[YYSX]}','${body[TYSX]}','${body[BS]}' " +
                                 "where NOT EXISTS ( SELECT * FROM centralbase.cb_pc_pro_wellbore_status_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}')"))
+                        .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-2->  centralbase.cb_pc_pro_wellbore_status_daily insert data failed")
+                        .log("${header.date}"+" routeId:insert-statusDaily ->  centralbase.cb_pc_pro_wellbore_status_daily insert data failed")
                         .end();
 
-                from("timer:mytimer5?period=3600000")
-                        .routeId("oracle-5")
+
+                from("timer:mytimer-update-statusDaily-DYM?period=3600000")
+                        .routeId("update-statusDaily-DYM")
                         .setHeader("date", constant(getDate() + " 00:00:00"))
-                        //三个月之内dym不为空的数据
-                        //.setBody(simple("SELECT distinct jh,rq,dym FROM DBA01 WHERE  rq between to_date(TO_CHAR(ADD_MONTHS(SYSDATE,-3),'yyyy-MM-dd'),'yyyy-MM-dd')and to_date(TO_CHAR(SYSDATE,'yyyy-MM-dd'),'yyyy-MM-dd') and dym is not null;"))
                         .setBody(simple("SELECT  distinct jh,max(rq),dym FROM DBA01 WHERE dym is not null group by jh,dym"))
                         .to("jdbc:oracle")
                         .split(body())
-                        .doTry()
                         .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}' "))
+                        .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-5->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
+                        .log("${header.date}"+" routeId:update-statusDaily-DYM ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
                         .end();
 
-                from("timer:mytimer7?period=3600000")
-                        .routeId("oracle-7")
+                from("timer:mytimer-update-statusDaily-submergenceDepth?period=3600000")
+                        .routeId("update-statusDaily-submergenceDepth")
                         .setHeader("date", constant(getDate() + " 00:00:00"))
-                        //.setBody(simple("SELECT  distinct jh,max(rq),dym FROM DBA01 WHERE dym is not null group by jh,dym"))
                         .setBody(simple("select well_id,prod_date,start_pump_liq_level,pump_depth from centralbase.cb_pc_pro_wellbore_status_daily where prod_date = '${header.date}' "))
                         .to("jdbc:centralbase")
                         .split(body()).process(exchange -> {
@@ -227,15 +201,36 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                                 aRow.put("submergence_depth",cmd1);
                             }
                 })
-                        .doTry()
                         .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set submergence_depth = '${body[submergence_depth]}' where well_id = '${body[well_id]}' and prod_date  = '${body[prod_date]}'"))
+                        .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-7->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
+                        .log("${header.date}"+" routeId:update-statusDaily-submergenceDepth ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
                         .end();
 
-                from("timer:mytimer3?period=3600000")
-                        .routeId("oracle-3")
+
+                from("timer:mytimer-update-statusDaily-oil_nozzle?period=3600000")
+                        .routeId("update-statusDaily-oil_nozzle")
+                        .setHeader("date", constant(getDate() + " 00:00:00"))
+                        .setBody(simple("select distinct rn.well_id,cb.prod_date,rn.pump_diameter  from centralbase.cb_temp_well_mech_runtime rn ,centralbase.cb_pc_pro_wellbore_status_daily cb where cb.well_id = rn.well_id " +
+                                "and cb.prod_date = '${header.date}' "))
+                        .to("jdbc:centralbase")//.log("${body}")
+                        .split(body()).process(exchange -> {
+                            Message in = exchange.getIn();
+                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                            aRow.putIfAbsent("pump_diameter", "0.0");
+                        })
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '${body[pump_diameter]}' where well_id ='${body[well_id]}' and prod_date='${body[prod_date]}' "))
+                        .doTry()
+                        .to("jdbc:centralbase")
+                        .doCatch(Exception.class)
+                        .log("${header.date}"+" routeId:update-statusDaily-oil_nozzle ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
+                        .end();
+
+
+
+                from("timer:mytimer-insert-volDaily?period=3600000")
+                        .routeId("insert-volDaily")
                         .setHeader("date", constant(getDate()))
                         .setBody(simple("select distinct  jh,rq,scsj, rcyl1,rcyl,rcql,hs, bz from DBA01 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
                         .to("jdbc:oracle")
@@ -282,26 +277,23 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         aRow.put("YMD",0.85);
                     }
                 })
-                        .doTry()
                         .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_vol_daily(well_id,prod_date,prod_time,liq_prod_daily,oil_prod_daily,gas_prod_daily,water_cut,remarks,gas_oil_ratio,water_prod_daily,water_gas_ratio,surface_crude_water_density,surface_crude_oil_density) " +
                                 "select '${body[JH]}','${body[RQ]}','${body[SCSJ]}','${body[RCYL1]}','${body[RCYL]}','${body[RCQL]}','${body[HS]}','${body[BZ]}','${body[QYB]}','${body[RCSL]}','${body[SQB]}','${body[SMD]}','${body[YMD]}' " +
                                 "where NOT EXISTS ( SELECT * FROM centralbase.cb_pc_pro_wellbore_vol_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}' )"))
+                        .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-3->  centralbase.cb_pc_pro_wellbore_vol_daily insert data failed")
-                        .doTry()
-                        .setBody(simple("routeId:oracle-3->  update centralbase.cb_pc_pro_wellbore_vol_daily set  water_prod_daily =null where water_prod_daily = -1 and prod_date = '${header.date}' "))
+                        .log("${header.date}"+" routeId:insert-volDaily ->  centralbase.cb_pc_pro_wellbore_vol_daily insert data failed")
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_vol_daily set  water_prod_daily =null where water_prod_daily = -1 and prod_date = '${header.date}' "))
                         .to("jdbc:centralbase")
-                        .setBody(simple("routeId:oracle-3->  update centralbase.cb_pc_pro_wellbore_vol_daily set  gas_oil_ratio =null where gas_oil_ratio = -1 and prod_date = '${header.date}' "))
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_vol_daily set  gas_oil_ratio =null where gas_oil_ratio = -1 and prod_date = '${header.date}' "))
                         .to("jdbc:centralbase")
-                        .setBody(simple(" routeId:oracle-3->  update centralbase.cb_pc_pro_wellbore_vol_daily set  water_gas_ratio =null where water_gas_ratio = -1 and prod_date = '${header.date}' "))
+                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_vol_daily set  water_gas_ratio =null where water_gas_ratio = -1 and prod_date = '${header.date}' "))
                         .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-3->  centralbase.cb_pc_pro_wellbore_vol_daily update data failed")
                         .end();
 
-                from("timer:mytimer12?period=3600000")
-                        .routeId("oracle-12")
+                from("timer:mytimer-update-volDaily-liq_prod_daily?period=3600000")
+                        .routeId("update-volDaily-liq_prod_daily")
                         .setHeader("date", constant(getDate()))
                         .setBody(simple("select distinct  jh,rq,scsj, rcyl1,rcyl,rcql,hs, bz from DBA01 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
                         .to("jdbc:oracle")
@@ -348,193 +340,18 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                                 aRow.put("YMD",0.85);
                             }
                         })
-                        .doTry()
                         .setBody(simple("update centralbase.cb_pc_pro_wellbore_vol_daily set prod_time = '${body[SCSJ]}' ,liq_prod_daily='${body[RCYL1]}' ,oil_prod_daily ='${body[RCYL]}' ,gas_prod_daily ='${body[RCQL]}' ,water_cut='${body[HS]}' ,remarks='${body[BZ]}' ,gas_oil_ratio='${body[QYB]}' ,water_prod_daily='${body[RCSL]}' ,water_gas_ratio='${body[SQB]}',surface_crude_water_density='${body[SMD]}',surface_crude_oil_density= '${body[YMD]}' " +
                                 "where well_id = '${body[JH]}' and prod_date ='${body[RQ]}' "))
-                        .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-12->  centralbase.cb_pc_pro_wellbore_vol_daily update data failed")
-                        .end();
-
-                from("timer:mytimer4?period=3600000")
-                        .routeId("oracle-4")
-                        .setHeader("date", constant(getDate()))
-                        .setBody(simple("select distinct jh,rq,dym,jy,ly,bj,bs,bx,zs,cc,cs,blx,dl from DBA01 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null  "))
-                        .to("jdbc:oracle")
-                        .split(body()).process(exchange -> {
-                    Message in = exchange.getIn();
-                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    if (aRow.get("JY") == null) aRow.put("JY", "0.0");
-                    if (aRow.get("LY") == null) aRow.put("LY", "0.0");
-                    if (aRow.get("BJ") == null) aRow.put("BJ", "0.0");
-                    if (aRow.get("BS") == null) aRow.put("BS", "0.0");
-                    if (aRow.get("BX") == null) aRow.put("BX", "0.0");
-                    if (aRow.get("ZS") == null) aRow.put("ZS", "0.0");
-                    if (aRow.get("CC") == null) aRow.put("CC", "0.0");
-                    if (aRow.get("CS") == null) aRow.put("CS", "0.0");
-                    if (aRow.get("BLX") == null) aRow.put("BLX", "");
-                    if (aRow.get("DL") == null) aRow.put("DL", "0.0");
-                })
                         .doTry()
-                        .setBody(simple("insert into centralbase.cb_temp_well_mech_daily(well_id,prod_date,static_pressure,flow_pres,pump_diameter,pump_depth,pump_efficiency,rotate_frequency,stroke_length,stroke_frequency,pump_type,elec_frequency) " +
-                                "select '${body[JH]}','${body[RQ]}','${body[JY]}','${body[LY]}','${body[BJ]}','${body[BS]}','${body[BX]}','${body[ZS]}','${body[CC]}','${body[CS]}','${body[BLX]}','${body[DL]}' " +
-                                "where NOT EXISTS ( SELECT * FROM centralbase.cb_temp_well_mech_daily WHERE well_id = '${body[JH]}' and  prod_date = '${body[RQ]}' )"))
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-4->  centralbase.cb_temp_well_mech_daily insert data failed")
-                        .end();
-
-                from("timer:mytimer6?period=3600000")
-                        .routeId("oracle-6")
-                        .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,bj"))
-                        .to("jdbc:oracle")
-                        .split(body()).process(exchange -> {
-                            Message in = exchange.getIn();
-                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                            if (aRow.get("BJ") ==null || aRow.get("BJ").equals("")) aRow.put("BJ","0.0");
-                        })
-                        .setBody(simple("update centralbase.cb_temp_well_mech_runtime set pump_diameter = '${body[BJ]}' where well_id = '${body[JH]}' "))
-                        .to("jdbc:centralbase")
-                        .log("${header.date}"+" routeId:oracle-6->  centralbase.cb_temp_well_mech_runtime update data failed")
-                        .end();
-
-                from("timer:mytimer8?period=3600000")
-                        .routeId("jdbc-gtsj-?")
-                        .setBody(simple("select max(prod_date) from centralbase.cb_temp_well_mech_runtime "))
-                        .to("jdbc:centralbase")
-                        .split(body())
-                        .setHeader("date", simple("${body[max]}"))
-                        .setBody(simple("select distinct well_name,dyna_create_time,check_date,displacement,disp_load,stroke,frequency,susp_max_load,susp_min_load from public.pc_fd_pumpjack_dyna_dia_t where   dyna_create_time > '${header.date}' "))
-                        .to("jdbc:gtsj")
-                        .split(body()).process(exchange -> {
-                    Message in = exchange.getIn();
-                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    String prod_date = aRow.get("dyna_create_time").toString().split("\\+")[0];
-                    aRow.put("dyna_create_time", prod_date);
-                    if (aRow.get("displacement") != null && !aRow.get("displacement").equals("") && aRow.get("disp_load") != null && !aRow.get("disp_load").equals("")) {
-                        String[] displacements = aRow.get("displacement").toString().split(";");//10 四舍五入
-                        //String[] displacements = wy(displacementsOld);
-                        String[] disp_loads = aRow.get("disp_load").toString().split(";");
-                        Double susp_max_load = max(disp_loads);
-                        Double susp_min_load = min(disp_loads);
-                        String sgt = "";
-                        for (int i = 0; i < displacements.length; i++) {
-                            sgt = sgt + displacements[i] + "," + disp_loads[i] + ",";
-                        }
-                        String[] s = sgt.split(",");
-                        String w = "";
-                        for (int i = 0; i < s.length; i++) {
-                            w += new BigDecimal(Math.round(Double.parseDouble(s[i]) * 100)).stripTrailingZeros().toPlainString() + ",";
-                        }
-                        Double[][] doubles = SGTUtil.encodeToDoubleArray(w);
-                        aRow.put("sgt", SGTUtil.encodeToString(doubles));
-                        aRow.put("susp_max_load",susp_max_load);
-                        aRow.put("susp_min_load",susp_min_load);
-                    }
-                    if (aRow.get("stroke") == null) aRow.put("stroke", "0.0");
-                    if (aRow.get("frequency") == null) aRow.put("frequency", "0.0");
-                    if (aRow.get("susp_max_load") == null) aRow.put("susp_max_load", "0.0");
-                    if (aRow.get("susp_min_load") == null) aRow.put("susp_min_load", "0.0");
-                    if (aRow.get("frequency") != null){
-                        BigDecimal bd=new BigDecimal(aRow.get("frequency").toString());
-                        double frequency=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
-                        aRow.put("frequency",frequency);
-                    }
-                    if (aRow.get("stroke") != null){
-                        double stroke1 = Double.parseDouble(aRow.get("stroke").toString());
-                        BigDecimal bd=new BigDecimal(stroke1);
-                        double stroke=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
-                        aRow.put("stroke",stroke);
-                    }
-                })
-                        .doTry()
-                        .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,stroke_length,stroke_frequency,susp_max_load,susp_min_load,sgt) " +
-                                "select '${body[well_name]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}' " +
-                                "where NOT EXISTS (SELECT * FROM centralbase.cb_temp_well_mech_runtime WHERE well_id = '${body[well_name]}' and  prod_date = '${body[dyna_create_time]}' )"))
-                        .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:jdbc-gtsj-?->  centralbase.cb_temp_well_mech_runtime insert data failed ${body}")
-                        .end();
-
-                  from("timer:mytimer9?period=3600000")
-                        .routeId("centralbase-2")
-                        .setHeader("date", constant(getDate() + " 00:00:00"))
-                        .setBody(simple("select distinct rn.well_id,cb.prod_date,rn.pump_diameter  from centralbase.cb_temp_well_mech_runtime rn ,centralbase.cb_pc_pro_wellbore_status_daily cb where cb.well_id = rn.well_id\n" +
-                                "and cb.prod_date = '${header.date}' "))
-                        .to("jdbc:centralbase")//.log("${body}")
-                        .split(body()).process(exchange -> {
-                            Message in = exchange.getIn();
-                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                            aRow.putIfAbsent("pump_diameter", "0.0");
-                })
-                        .doTry()
-                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '${body[pump_diameter]}' where well_id ='${body[well_id]}' and prod_date='${body[prod_date]}' "))
-                        .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:centralbase-2->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
-                        .end();
-
-               from("timer:mytimer10?period=3600000")
-                        .routeId("centralbase-3")
-                        .setHeader("date", constant(getDate()))
-                        .setBody(simple("select well_id,avg(stroke_length) stroke_length  ,avg(stroke_frequency) stroke_frequency from centralbase.cb_temp_well_mech_runtime where prod_date::date='${header.date}' group by well_id"))
-                        .to("jdbc:centralbase")//.log("${body}")
-                        .split(body()).process(exchange -> {
-                    Message in = exchange.getIn();
-                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    if (aRow.get("stroke_length")!=null && aRow.get("stroke_frequency")!=null){
-                        double stroke_length=Double.parseDouble(aRow.get("stroke_length").toString());
-                        double stroke_frequency=Double.parseDouble(aRow.get("stroke_frequency").toString());
-                        BigDecimal bd=new BigDecimal(stroke_length);
-                        double stroke_lengt1=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
-                        BigDecimal bd1=new BigDecimal(stroke_frequency);
-                        double stroke_frequency1=bd1.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
-                        aRow.put("strokeLength",stroke_lengt1);
-                        aRow.put("strokeFrequency",stroke_frequency1);
-                    }
-                })
-                        .doTry()
-                        .setBody(simple("update centralbase.cb_temp_well_mech_daily set stroke_length='${body[strokeLength]}' ,stroke_frequency ='${body[strokeFrequency]}' where well_id = '${body[well_id]}' and prod_date::date='${header.date}' "))
-                        .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:centralbase-3->  centralbase.cb_temp_well_mech_daily update data failed")
+                        .log("${header.date}"+" routeId:update-volDaily-liq_prod_daily ->  centralbase.cb_pc_pro_wellbore_vol_daily update data failed")
                         .end();
+            };
+        };
 
-                from("timer:mytimer11?period=3600000")
-                        .routeId("centralbase-1")
-                        .setBody(simple("select so.well_id,so.well_common_name,so.org_id,ti.prod_date,ti.stroke_frequency,ti.stroke_length,ti.sgt from centralbase.cb_temp_well_mech_runtime ti, centralbase.cb_cd_well_source so where ti.well_id = so.well_id and ti.prod_date = (select max(prod_date) from centralbase.cb_temp_well_mech_runtime) "))
-                        .to("jdbc:centralbase").process(exchange -> {
-                            sendMsgRunTime=0;
-                            DefaultMsgSendSuccessCallBack.count.set(0);
-                        })
-                        .split(body())
-                        .doTry()
-                        .process(exchange -> {
-                            Message in = exchange.getIn();
-                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                            String wellName =aRow.get("well_common_name").toString();
-                            String wellId =aRow.get("well_id").toString();
-                            String orgId = aRow.get("org_id").toString();
-                            String prodDate = aRow.get("prod_date").toString().substring(0,19);
-                            Double strokeLength = Double.valueOf(aRow.get("stroke_length").toString());
-                            Double strokeFrequency = Double.valueOf(aRow.get("stroke_frequency").toString());
-                            String sgt = aRow.get("sgt").toString();
-                            if (sgt == null || sgt.length() ==0){
-                                sgt = "0,0";
-                            }
-                            DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), strokeLength, strokeFrequency);
-                            sendMsgRunTime++;
-                            System.out.println("sendMsgRunTime = " + sendMsgRunTime);
-                            producer.send(diagnoseMsg);
-                        })
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" rocketMQ send data failed")
-                        .end();
 
 
-            };
-        };
 
     }
 }