正常使用版本

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

@@ -98,9 +98,7 @@ public class GtController {
     @GetMapping("/getGTSJ")
     @ApiOperation(value = "从实时的机采数据生产的表中获取数据放入到 Runtime 表中")
     public JSONObject getGTSJ(@RequestBody GTBody gtBody){
-
         String startDate = gtBody.getStartDate();
-        String endDate = gtBody.getEndDate();
         JSONArray wellList = gtBody.getWellList();
         JSONObject jsonObject = new JSONObject();
         jdbcTemplate = new JdbcTemplate(dataSourceOfGTSJ);
@@ -108,54 +106,29 @@ public class GtController {
         int pageSize = 5000;
         int sumData = 0;
         try {
-            for (int i = 0; i < wellList.size(); i++) {
-                String sql = String.format("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 > '%s' and well_name = '%s' offset %d limit %d ", startDate, wellList.get(i));
-                List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
-                for (Map<String, Object> map : list) {
-                    String prod_date = map.get("dyna_create_time").toString().split("\\+")[0];
-                    map.put("dyna_create_time", prod_date);
-                    if (map.get("displacement") != null && !map.get("displacement").equals("") && map.get("disp_load") != null && !map.get("disp_load").equals("")) {
-                        String[] displacements = map.get("displacement").toString().split(";");//10 四舍五入
-                        String[] disp_loads = map.get("disp_load").toString().split(";");
-                        Double susp_max_load = max(disp_loads);
-                        Double susp_min_load = min(disp_loads);
-                        String sgt = "";
-                        for (int y = 0; y < displacements.length; y++) {
-                            sgt = sgt + displacements[y] + "," + disp_loads[y] + ",";
-                        }
-                        String[] s = sgt.split(",");
-                        String w = "";
-                        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));
-                        map.put("susp_max_load",susp_max_load);
-                        map.put("susp_min_load",susp_min_load);
-                    }
-                    if (map.get("stroke") == null) map.put("stroke", "0.0");
-                    if (map.get("frequency") == null) map.put("frequency", "0.0");
-                    if (map.get("susp_max_load") == null) map.put("susp_max_load", "0.0");
-                    if (map.get("susp_min_load") == null) map.put("susp_min_load", "0.0");
-                    if (map.get("frequency") != null){
-                        BigDecimal bd=new BigDecimal(map.get("frequency").toString());
-                        double frequency=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
-                        map.put("frequency",frequency);
-                    }
-                    if (map.get("stroke") != null){
-                        double stroke1 = Double.parseDouble(map.get("stroke").toString());
-                        BigDecimal bd=new BigDecimal(stroke1);
-                        double stroke=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
-                        map.put("stroke",stroke);
-                    }
-                    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) " +
-                            "values (?,?,?,?,?,?,?)",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();
-                if (list.size()< pageSize){
-                    break;
+            if (wellList.isEmpty() || wellList.size() ==0 ){
+                while (true){
+                    String sql = String.format("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 > '%s'  limit %d offset %d ", startDate,curPage,pageSize);
+                    List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
+                    sql = String.format("select count(1) from public.pc_fd_pumpjack_dyna_dia_t where   dyna_create_time > '%s' ", startDate );
+                    Integer count = jdbcTemplate.queryForObject(sql, Integer.class);
+                    int pageNum = (count % pageSize ==0 ) ?(count / pageSize) : (count/pageSize)+1;
+                    insertDataToRunTime(list);
+                    sumData += list.size();
+                    curPage ++;
+                    if (curPage>pageNum) break;
                 }
-                curPage ++ ;
+            }else {
+              for (int i = 0; i < wellList.size(); i++) {
+                  String sql = String.format("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 > '%s' and well_name = '%s'  limit %d offset %d ", startDate, wellList.get(i),curPage,pageSize);
+                  List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
+                  insertDataToRunTime(list);
+                  sumData += list.size();
+                  if (list.size()< pageSize){
+                      break;
+                  }
+                  curPage ++ ;
+              }
             }
         }catch (Exception e){
             jsonObject.put("error",e.getMessage());
@@ -163,8 +136,50 @@ public class GtController {
             jsonObject.put("sumData",sumData);
         }
         return    jsonObject;
-         }
-
+}
+    public void insertDataToRunTime(List<Map<String, Object>> list){
+        for (Map<String, Object> map : list) {
+            String prod_date = map.get("dyna_create_time").toString().split("\\+")[0];
+            map.put("dyna_create_time", prod_date);
+            if (map.get("displacement") != null && !map.get("displacement").equals("") && map.get("disp_load") != null && !map.get("disp_load").equals("")) {
+                String[] displacements = map.get("displacement").toString().split(";");//10 四舍五入
+                String[] disp_loads = map.get("disp_load").toString().split(";");
+                Double susp_max_load = max(disp_loads);
+                Double susp_min_load = min(disp_loads);
+                String sgt = "";
+                for (int y = 0; y < displacements.length; y++) {
+                    sgt = sgt + displacements[y] + "," + disp_loads[y] + ",";
+                }
+                String[] s = sgt.split(",");
+                String w = "";
+                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));
+                map.put("susp_max_load",susp_max_load);
+                map.put("susp_min_load",susp_min_load);
+            }
+            if (map.get("stroke") == null) map.put("stroke", "0.0");
+            if (map.get("frequency") == null) map.put("frequency", "0.0");
+            if (map.get("susp_max_load") == null) map.put("susp_max_load", "0.0");
+            if (map.get("susp_min_load") == null) map.put("susp_min_load", "0.0");
+            if (map.get("frequency") != null){
+                BigDecimal bd=new BigDecimal(map.get("frequency").toString());
+                double frequency=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                map.put("frequency",frequency);
+            }
+            if (map.get("stroke") != null){
+                double stroke1 = Double.parseDouble(map.get("stroke").toString());
+                BigDecimal bd=new BigDecimal(stroke1);
+                double stroke=bd.setScale(1,BigDecimal.ROUND_HALF_UP).doubleValue();
+                map.put("stroke",stroke);
+            }
+            jdbcTemplate = new JdbcTemplate(dataSource);
+            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) " +
+                    "values (?,?,?,?,?,?,?)",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"));
+        }
+   }
 
     public Double min(String[] strings){
         double[] doubles = new double[strings.length];

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

@@ -97,16 +97,14 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         .end();*/
 
                 from("timer:mytimer-insert-runtime?period=1800000")
-                        .routeId("insert-runtimeAndSendToMQ")
+                        .routeId("insert-runtime")
                         .setBody(simple("select max(prod_date)  from centralbase.cb_temp_well_mech_runtime "))
                         .to("jdbc:centralbase")
+                        .log("max(prod_date) ${body}")
                         .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("${header.date}" + " routeId:insert-runtimeAndSendToMQ -> select runTime data failed")
                         .process(exchange -> {
                             sendMsgRunTime = 0;
                         })
@@ -114,6 +112,7 @@ public class CamelJDBCCofRealTimeConfiguration  {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);
                     String prod_date = aRow.get("dyna_create_time").toString().split("\\+")[0];
+                    System.out.println("prod_date---"+prod_date);
                     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 四舍五入
@@ -156,10 +155,7 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         .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-SendToMQ?period=180000")
@@ -178,9 +174,6 @@ public class CamelJDBCCofRealTimeConfiguration  {
                             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);
                             producer.send((MessageBody) diagnoseMsg);
                         })

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

@@ -300,6 +300,7 @@ public class CamelJDBCConfiguration {
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
                         .log("${header.date}"+" routeId:insert-volDaily ->  centralbase.cb_pc_pro_wellbore_vol_daily insert data failed")
+                        .endDoTry()
                         .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("update centralbase.cb_pc_pro_wellbore_vol_daily set  gas_oil_ratio =null where gas_oil_ratio = -1 and prod_date = '${header.date}' "))