新的稳定版

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

@@ -4,12 +4,10 @@ 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;
 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;
@@ -17,8 +15,6 @@ 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
@@ -58,18 +54,15 @@ public class CamelJDBCCofRealTimeConfiguration  {
      * 异步发送 RunTime 消息到MQ中
      * */
    // @Async
-    public long sendDataToRocketMQ(String wellName,String wellId,String prodDate,String stroke_length ,String stroke_frequency,String sgt){
+    public void sendDataToRocketMQ(String wellName, String wellId, String prodDate, Double stroke_length , Double 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);
+        DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), stroke_length, stroke_frequency);
         sendMsgRunTime++;
         producer.send(diagnoseMsg);
-        return sendMsgRunTime;
     }
 
     @Resource(name = "diagnoseMessageProducer")
@@ -149,7 +142,13 @@ public class CamelJDBCCofRealTimeConfiguration  {
                     }
                            // 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"));
-
+                            String wellName =aRow.get("well_name").toString();
+                            String wellId =aRow.get("well_name").toString();
+                            String prodDate = aRow.get("dyna_create_time").toString().substring(0,19);
+                            Double strokeLength = Double.valueOf(aRow.get("stroke").toString());
+                            Double strokeFrequency = Double.valueOf(aRow.get("frequency").toString());
+                            String sgt = aRow.get("sgt").toString();
+                            sendDataToRocketMQ(wellName,wellId,prodDate,strokeLength,strokeFrequency,sgt);
                         })
                         .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]}' " +
@@ -157,7 +156,7 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         .to("jdbc:centralbase")
                         .end();
 
-                from("timer:mytimer-SendToMQ?period=180000")
+                /*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")
@@ -178,7 +177,8 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         })
                         .doCatch(Exception.class)
                         .log("${header.date}"+" rocketMQ send data failed")
-                        .end();
+                        .endDoTry()
+                        .end();*/
 
                from("timer:mytimer-update-avg-mech_daily?period=3600000")
                         .routeId("update-avg-mech_daily")