修改sendDataToMQ logic and add select a2 and gtsj 接口

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

@@ -14,8 +14,10 @@ import org.springframework.web.bind.annotation.RestController;
 import javax.annotation.Resource;
 import javax.sql.DataSource;
 import java.math.BigDecimal;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * class name: BaseDataController
@@ -36,8 +38,11 @@ public class BaseDataController {
     @Resource(name = "centralbase")
     DataSource baseDataSource;
 
+    @Resource(name = "gtsj")
+    DataSource gtsjDataSource;
+
     @PostMapping("/saveDataToStatusDaily")
-    @ApiOperation("从 A2 获取 数据插入到 centrlBase-StatusDaily 中")
+    @ApiOperation("从 A2 获取 数据插入到 centrlBase-StatusDaily 中,只需要传入查询时间")
     public JSONObject saveDataToStatusDaily(@RequestBody BaseDataBody baseDataBody) {
         JSONObject jsonObject = new JSONObject();
         int insertCount = 0;
@@ -122,7 +127,7 @@ public class BaseDataController {
 
 
     @PostMapping("/saveDataToVolDaily")
-    @ApiOperation("从 A2 获取 数据插入到 centrlBase-VolDaily 中")
+    @ApiOperation("从 A2 获取 数据插入到 centrlBase-VolDaily 中,只需要传入查询时间")
     public JSONObject saveVolDaily(@RequestBody BaseDataBody baseDataBody) {
         JSONObject jsonObject = new JSONObject();
         int insertCount = 0;
@@ -198,4 +203,63 @@ public class BaseDataController {
         jdbcTemplate = new JdbcTemplate(baseDataSource);
         return jdbcTemplate.update(sql);
     }
+
+    @PostMapping("/getNeedRunTimeData")
+    @ApiOperation("查询天安的数据库 只需要传过来 需要查询的sql 就行")
+    public JSONObject getNeedRunTimeData(@RequestBody BaseDataBody baseDataBody){
+        JSONObject jsonObject = new JSONObject();
+        jdbcTemplate = new JdbcTemplate(gtsjDataSource);
+        try{
+            List<Map<String, Object>> list = jdbcTemplate.queryForList(baseDataBody.getSql());
+            jsonObject.put("RunTimedata",list);
+        }catch (Exception e){
+            jsonObject.put("error",e.getMessage());
+        }
+         return jsonObject;
+    }
+
+    @PostMapping("/getNeedA2Data")
+    @ApiOperation("查询A2数据库 只需要传过来 需要查询的sql 就行")
+    public JSONObject getNeedA2Data(@RequestBody BaseDataBody baseDataBody){
+        JSONObject jsonObject = new JSONObject();
+        jdbcTemplate = new JdbcTemplate(oracleDataSource);
+        try{
+            List<Map<String, Object>> list = jdbcTemplate.queryForList(baseDataBody.getSql());
+            jsonObject.put("A2data",list);
+        }catch (Exception e){
+            jsonObject.put("error",e.getMessage());
+        }
+        return jsonObject;
+    }
+
+    @PostMapping("/getA2WellSorceData")
+    @ApiOperation("查询A2数据库 并且和 wellSource 做对比 只填写需要查询A2Data 的时间就行")
+    public JSONObject getA2WellSorceData(@RequestBody BaseDataBody baseDataBody){
+        JSONObject jsonObject = new JSONObject();
+        jdbcTemplate = new JdbcTemplate(oracleDataSource);
+        try{
+            String sql = "select  distinct jh  from DBA01 where rq  = to_date('"+baseDataBody.getDate()+"','yyyy-MM-dd') and qyrq is not null ";
+            List<Map<String, Object>>  A2DataList = jdbcTemplate.queryForList(sql);
+            List<Map<String, Object>>  wellSourceList = getWellSource();
+            Object collect = A2DataList.stream().filter(item -> !wellSourceList.contains(item)).collect(Collectors.toList());
+            Object collect1 = wellSourceList.stream().filter(item -> !A2DataList.contains(item)).collect(Collectors.toList());
+            jsonObject.put("wellSource 中不存在 A2 的井号为",collect);
+            jsonObject.put("A2 中不存在 wellSource 的井号为",collect1);
+        }catch (Exception e){
+            jsonObject.put("error",e.getMessage());
+        }
+        return jsonObject;
+    }
+
+    public List<Map<String, Object>>  getWellSource(){
+        List<Map<String, Object>>  list = new ArrayList<>();
+        try{
+            jdbcTemplate = new JdbcTemplate(baseDataSource);
+            String sql = " select distinct well_id from centralbase.cb_cd_well_source ";
+            list = jdbcTemplate.queryForList(sql);
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+        return list;
+    }
 }

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

@@ -18,8 +18,11 @@ public class BaseDataBody {
 
     @ApiModelProperty(value = "时间")
     public String date;
+
+    @ApiModelProperty(value = "sql语句")
+    public String sql;
     /*@ApiModelProperty(value = "结束时间")
     public String endDate;*/
-    @ApiModelProperty(value = "井号集合")
-    public JSONArray wellList;
+    /*@ApiModelProperty(value = "井号集合")
+    public JSONArray wellList;*/
 }

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

@@ -1,6 +1,7 @@
 package com.gct.tools.etlcamelhuge.routeconfig;
 
 import com.gct.common.util.SGTUtil;
+import com.gct.tools.etlcamelhuge.MQ.MessageBody;
 import com.gct.tools.etlcamelhuge.MQ.MessageProducer;
 import com.gct.tools.etlcamelhuge.entity.DiagnoseMsg;
 import lombok.Data;
@@ -56,7 +57,7 @@ public class CamelJDBCCofRealTimeConfiguration  {
     /***
      * 异步发送 RunTime 消息到MQ中
      * */
-    @Async
+   // @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);
@@ -95,7 +96,7 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         .log("${header.date}"+" routeId:update-runTime -> centralbase.cb_temp_well_mech_runtime update data failed")
                         .end();*/
 
-                from("timer:mytimer-insert-runtimeAndSendToMQ?period=1800000")
+                from("timer:mytimer-insert-runtime?period=1800000")
                         .routeId("insert-runtimeAndSendToMQ")
                         .setBody(simple("select max(prod_date)  from centralbase.cb_temp_well_mech_runtime "))
                         .to("jdbc:centralbase")
@@ -148,8 +149,8 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         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"));
+                           // 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) " +
@@ -161,6 +162,32 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         .log("${header.date}"+" routeId:insert-runtimeAndSendToMQ ->  centralbase.cb_temp_well_mech_runtime insert data failed ${body}")
                         .end();
 
+                from("timer:mytimer-SendToMQ?period=180000")
+                        .routeId("SendToMQ")
+                        .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")
+                        .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);
+                            producer.send((MessageBody) diagnoseMsg);
+                        })
+                        .doCatch(Exception.class)
+                        .log("${header.date}"+" rocketMQ send data failed")
+                        .end();
+
                from("timer:mytimer-update-avg-mech_daily?period=3600000")
                         .routeId("update-avg-mech_daily")
                         .setHeader("date", constant(getDate()))

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

@@ -148,6 +148,7 @@ public class CamelJDBCConfiguration {
                         .log("${header.date}"+" routeId:oracle-1->  centralbase.cb_pc_organization update data failed")
                         .end();*/
 
+
                 from("timer:mytimer-insert-statusDaily?period=3600000")
                         .routeId("insert-statusDaily")
                         .setHeader("date", constant(getDate()))