Merge remote-tracking branch 'origin/zygs-1024' into zygs-1024

# Conflicts:
#	src/main/java/com/gct/tools/etlcamelhuge/routeconfig/CamelJDBCConfiguration.java

src/main/java/com/gct/tools/etlcamelhuge/MQ/DiagnoseMessageProducer.java

@@ -49,6 +49,7 @@ public class DiagnoseMessageProducer implements MessageProducer {
             return;
         }
         DiagnoseMsg actualMsg = (DiagnoseMsg) msgBody;
+        System.out.println(properties);
         Message msg = new Message(properties.getTopic(), properties.getTags(), actualMsg.getWellId(), JSONObject.toJSONString(actualMsg).getBytes(StandardCharsets.UTF_8));
         sendDefault(producer, msg, (MessageBody) actualMsg, failCallBack, successCallBack,0);
     }

src/main/java/com/gct/tools/etlcamelhuge/MQ/MessageProducer.java

@@ -1,6 +1,7 @@
 package com.gct.tools.etlcamelhuge.MQ;
 
 import com.gct.tools.etlcamelhuge.entity.DiagnoseMsg;
+import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendCallback;
@@ -11,6 +12,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
 
 /**
  * class name: MessageProducer
@@ -32,6 +34,7 @@ public interface MessageProducer {
     default void sendDefault(DefaultMQProducer producer, Message msg, MessageBody actualMsg, SendFailCallBack failCallBack, SendSuccessCallBack successCallBack,int retryCount) {
         Logger log = LoggerFactory.getLogger(producer.getProducerGroup() + producer.getClass().getTypeName());
         try {
+
             producer.send(msg, new SendCallback() {
                 @Override
                 public void onSuccess(SendResult sendResult) {
@@ -49,6 +52,7 @@ public interface MessageProducer {
                 }
             });
         } catch (MQClientException | RemotingException | InterruptedException e) {
+            e.printStackTrace();
             log.error("Send msg failed,by cause:{},stack:{}", e.getMessage(), e.getStackTrace());
         }
     }

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

@@ -10,9 +10,13 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.util.StringUtils;
 
 import javax.annotation.Resource;
+import java.io.IOException;
+import java.io.Reader;
 import java.lang.management.MemoryMXBean;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
+import java.sql.Clob;
+import java.sql.SQLException;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
@@ -63,6 +67,7 @@ public class CamelJDBCCofRealTimeConfiguration {
     public void sendDataToRocketMQ(String wellName, String wellId, String prodDate, Double stroke_length, Double stroke_frequency, String sgt) {
         String orgId = "0";
         DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), stroke_length, stroke_frequency);
+        System.out.println(diagnoseMsg);
         sendMsgRunTime++;
         producer.send(diagnoseMsg);
     }
@@ -70,6 +75,25 @@ public class CamelJDBCCofRealTimeConfiguration {
     @Resource(name = "diagnoseMessageProducer")
     private MessageProducer producer;
 
+    private String clobToString(Clob clob){
+        StringBuilder sb = new StringBuilder();
+        if (clob != null) {
+            try (Reader reader = clob.getCharacterStream()) {
+                char[] buffer = new char[1024];
+                int bytesRead;
+                while ((bytesRead = reader.read(buffer)) != -1) {
+                    sb.append(buffer, 0, bytesRead);
+                }
+            } catch (SQLException e) {
+                throw new RuntimeException(e);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+        return sb.toString();
+
+    }
+
     @Bean
     public RouteBuilder routeBuilderWithRealTime() {
         return new RouteBuilder() {
@@ -82,10 +106,14 @@ public class CamelJDBCCofRealTimeConfiguration {
                         .split(body()).process(exchange -> {
                             HashMap body = exchange.getIn().getBody(HashMap.class);
                             exchange.getIn().setHeader("well_id", body.get("well_id"));
-                            exchange.getIn().setHeader("sgt_last_time", body.get("sgt_last_time"));
+//                            System.out.println(body.get("well_id"));
+                            if (Objects.isNull(body.get("sgt_last_time")))body.put("sgt_last_time"
+                                    ,LocalDateTime.now().minusHours(4).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+                            exchange.getIn().setHeader("sgt_last_time", body.get("sgt_last_time").toString().substring(0,19));
                         })
-                        .setBody(simple("select distinct well_name,dyna_create_time,check_date,displacement,disp_load,stroke,frequency,susp_max_load,susp_min_load" +
-                                " from DEFAULT_GONGTU where well_name='${header.well_id}' and dyna_create_time >= '${header.sgt_last_time}' limit 50 "))
+                        .setBody(simple("select WELL_ID,WELL_NAME,DYNA_CREATE_TIME,CHECK_DATE,DISPLACEMENT,DISP_LOAD,STROKE,FREQUENCY,SUSP_MAX_LOAD,SUSP_MIN_LOAD" +
+                                " from DEFAULT_GONGTU where CHECK_DATE >= to_date('${header.sgt_last_time}','yyyy-mm-dd hh24:mi:ss') " +
+                                "and well_id='${header.well_id}' and  rownum<50 "))
                         .to("jdbc:oracle")
                         .process(exchange -> {
                             sendMsgRunTime = 0;
@@ -93,14 +121,22 @@ public class CamelJDBCCofRealTimeConfiguration {
                         .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);
+                            String prod_date = aRow.get("DYNA_CREATE_TIME").toString().split("\\+")[0];
+                            aRow.put("DYNA_CREATE_TIME", prod_date);
                             aRow.put("sgt", "");
-                            if (!StringUtils.isEmpty(aRow.get("displacement")) && !StringUtils.isEmpty(aRow.get("disp_load"))) {
-                                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);
+                            //swap table error design column
+                            Object tempObj = aRow.get("STROKE");
+                            aRow.put("STROKE",aRow.get("FREQUENCY"));
+                            aRow.put("FREQUENCY",tempObj);
+                            if (!StringUtils.isEmpty(aRow.get("DISPLACEMENT")) && !StringUtils.isEmpty(aRow.get("DISP_LOAD"))) {
+                                String[] displacements =  clobToString((Clob) aRow.get("DISPLACEMENT")).split(",");//10 四舍五入
+                                for (int i = 0; i < displacements.length; i++) {
+                                    displacements[i] = Double.toString(Double.parseDouble(displacements[i])/10);
+                                }
+//                                System.out.println("disp="+Arrays.deepToString(displacements));
+                                String[] disp_loads = clobToString((Clob) aRow.get("DISP_LOAD")).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] + ",";
@@ -112,15 +148,15 @@ public class CamelJDBCCofRealTimeConfiguration {
                                 }
                                 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);
+                                aRow.put("SUSP_MAX_LOAD", SUSP_MAX_LOAD);
+                                aRow.put("SUSP_MIN_LOAD", SUSP_MIN_LOAD);
                             }
                             //对于位移没有数据,所有数据都在载荷中的特殊数据做特别处理
-                            if (StringUtils.isEmpty(aRow.get("displacement")) && !StringUtils.isEmpty(aRow.get("disp_load"))) {
-                                String disp_load = aRow.get("disp_load").toString().replaceAll(";", ",");
-                                Double[][] doubles = SGTUtil.encodeToDoubleArray(disp_load);
+                            if (StringUtils.isEmpty(aRow.get("DISPLACEMENT")) && !StringUtils.isEmpty(aRow.get("DISP_LOAD"))) {
+                                String DISP_LOAD = aRow.get("DISP_LOAD").toString().replaceAll(";", ",");
+                                Double[][] doubles = SGTUtil.encodeToDoubleArray(DISP_LOAD);
                                 aRow.put("sgt", SGTUtil.encodeToString(doubles));
-                                String[] split = disp_load.split(",");
+                                String[] split = DISP_LOAD.split(",");
                                 List<String> list = new ArrayList<>();
                                 for (int i = 0; i < split.length; i++) {
                                     if (i % 2 != 0) {
@@ -129,43 +165,44 @@ public class CamelJDBCCofRealTimeConfiguration {
                                     }
                                 }
                                 String[] loads = list.toArray(new String[0]);
-                                Double susp_max_load = null;
-                                Double susp_min_load = null;
+                                Double SUSP_MAX_LOAD = null;
+                                Double SUSP_MIN_LOAD = null;
                                 if (loads.length > 0) {
-                                    susp_max_load = max(loads);
-                                    susp_min_load = min(loads);
+                                    SUSP_MAX_LOAD = max(loads);
+                                    SUSP_MIN_LOAD = min(loads);
                                 }
-                                aRow.put("susp_max_load", susp_max_load);
-                                aRow.put("susp_min_load", susp_min_load);
+                                aRow.put("SUSP_MAX_LOAD", SUSP_MAX_LOAD);
+                                aRow.put("SUSP_MIN_LOAD", SUSP_MIN_LOAD);
                             }
-                            aRow.putIfAbsent("stroke", "0.0");
-                            aRow.putIfAbsent("frequency", "0.0");
-                            aRow.putIfAbsent("susp_max_load", "0.0");
-                            aRow.putIfAbsent("susp_min_load", "0.0");
-                            if (!StringUtils.isEmpty(aRow.get("frequency"))) {
-                                BigDecimal bd = new BigDecimal(aRow.get("frequency").toString());
-                                double frequency = bd.setScale(1, RoundingMode.HALF_UP).doubleValue();
-                                aRow.put("frequency", frequency);
+                            aRow.putIfAbsent("STROKE", "0.0");
+                            aRow.putIfAbsent("FREQUENCY", "0.0");
+                            aRow.putIfAbsent("SUSP_MAX_LOAD", "0.0");
+                            aRow.putIfAbsent("SUSP_MIN_LOAD", "0.0");
+                            if (!StringUtils.isEmpty(aRow.get("FREQUENCY"))) {
+                                BigDecimal bd = new BigDecimal(aRow.get("FREQUENCY").toString());
+                                double FREQUENCY = bd.setScale(1, RoundingMode.HALF_UP).doubleValue();
+                                aRow.put("FREQUENCY", FREQUENCY);
                             }
-                            if (!StringUtils.isEmpty(aRow.get("stroke"))) {
-                                double stroke1 = Double.parseDouble(aRow.get("stroke").toString());
+                            if (!StringUtils.isEmpty(aRow.get("STROKE"))) {
+                                double stroke1 = Double.parseDouble(aRow.get("STROKE").toString());
                                 BigDecimal bd = new BigDecimal(stroke1);
-                                double stroke = bd.setScale(1, RoundingMode.HALF_UP).doubleValue();
-                                aRow.put("stroke", stroke);
+                                double STROKE = bd.setScale(1, RoundingMode.HALF_UP).doubleValue();
+                                aRow.put("STROKE", STROKE);
                             }
-                            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 wellName = aRow.get("WELL_NAME").toString();
+                            String wellId = aRow.get("WELL_ID").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();
+//                            System.out.println(sgt);
                             in.setHeader("sgt_last_time", prodDate);
                             in.setHeader("well_id", wellId);
                             sendDataToRocketMQ(wellName, wellId, prodDate, strokeLength, strokeFrequency, sgt);
                         })
-                        .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,susp_max_load,susp_min_load,sgt) " +
-                                "select '${body[well_name]}','${body[dyna_create_time]}','${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]}' )"))
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,SUSP_MAX_LOAD,SUSP_MIN_LOAD,sgt,stroke_length,stroke_frequency) " +
+                                "select '${body[WELL_ID]}','${body[DYNA_CREATE_TIME]}','${body[SUSP_MAX_LOAD]}','${body[SUSP_MIN_LOAD]}','${body[sgt]}','${body[STROKE]}','${body[FREQUENCY]}' " +
+                                "where NOT EXISTS (SELECT * FROM centralbase.cb_temp_well_mech_runtime WHERE well_id = '${body[WELL_ID]}' and  prod_date = '${body[DYNA_CREATE_TIME]}' )"))
                         .to("jdbc:centralbase")
                         .setBody(simple("update centralbase.sys_access_well_control set sgt_last_time = '${header.sgt_last_time}' where well_id ='${header.well_id}' "))
                         .to("jdbc:centralbase")

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

@@ -11,6 +11,7 @@ import org.springframework.context.annotation.Configuration;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
+import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
@@ -79,14 +80,15 @@ public class CamelJDBCConfiguration {
                                 body.put("error_id", 1);
                                 body.put("updateTime", format);
                             }
+                            body.put("sgtLastTime",format);
 
                         })
                         //之前在数据导入时没有自动填写入最后更新时间,会导致新加入的数据无法自动导入
                         .setBody(simple("update centralbase.sys_access_well_control set well_common_name='${body[well_common_name]}'," +
-                                "org_id='${body[org_id]}',update_time = '${body[updateTime]}'::timestamp,sgt_last_time='${body[updateTime]}', " +
+                                "org_id='${body[org_id]}',update_time = '${body[updateTime]}'::timestamp,sgt_last_time='${body[sgtLastTime]}', " +
                                 "remarks ='${body[remarks]}' ,error_id ='${body[error_id]}' " +
-                                "where well_id ='${body[well_id]} and " +
-                                "(sgt_last_time is null or sgt_last_time<'${body[yesterday]}'::timestamp)"))
+                                "where well_id ='${body[well_id]}' and " +
+                                "(sgt_last_time is null or sgt_last_time::timestamp<'${body[yesterday]}'::timestamp)"))
                         .to("jdbc:centralbase")
                         .end();
 
@@ -135,10 +137,11 @@ public class CamelJDBCConfiguration {
                         });
                 setSysControlBody(statusDailyDYM)
                         .setBody(simple("select distinct WELL_ID , PROD_DATE , DYNAMIC_LIQ_LEVEL " +
-                                "from ZY_DXGW.PC_PRO_COM_DAILY " +
-                                "where  (WELL_ID,PROD_DATE) in" +
-                                " (SELECT  WELL_ID,max(PROD_DATE) PROD_DATE  FROM ZY_DXGW.PC_PRO_COM_DAILY" +
-                                "    WHERE DYNAMIC_LIQ_LEVEL is not null and WELL_ID='${header.well_id}' group by WELL_ID)"))
+                                "                                from ZY_DXGW.PC_PRO_COM_DAILY " +
+                                "                                where (WELL_ID,PROD_DATE) in " +
+                                "                                 (SELECT WELL_ID, max(PROD_DATE) PROD_DATE  FROM ZY_DXGW.PC_PRO_COM_DAILY " +
+                                "                                  where   PROD_DATE > TO_DATE('2024-01-01','yyyy-MM-dd')  " +
+                                "and WELL_ID='${header.well_id}' and DYNAMIC_LIQ_LEVEL is not null  group by WELL_ID)"))
                         .to("jdbc:oracle")
                         .split(body())
                         .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set start_pump_liq_level = '${body[DYNAMIC_LIQ_LEVEL]}'  " +
@@ -157,10 +160,11 @@ public class CamelJDBCConfiguration {
                         });
                 setSysControlBody(statusDailyBS)
                         .setBody(simple("select distinct WELL_ID , PROD_DATE , PUMP_DEPTH " +
-                                "from ZY_DXGW.PC_PRO_COM_DAILY " +
-                                "where  (WELL_ID,PROD_DATE) in" +
-                                " (SELECT  WELL_ID,max(PROD_DATE) PROD_DATE  FROM ZY_DXGW.PC_PRO_COM_DAILY" +
-                                "    WHERE PUMP_DEPTH is not null and WELL_ID='${header.well_id}' group by WELL_ID)"))
+                                "                                from ZY_DXGW.PC_PRO_COM_DAILY " +
+                                "                                where (WELL_ID,PROD_DATE) in " +
+                                "                                 (SELECT WELL_ID, max(PROD_DATE) PROD_DATE  FROM ZY_DXGW.PC_PRO_COM_DAILY " +
+                                "                                  where   PROD_DATE > TO_DATE('2024-01-01','yyyy-MM-dd')  " +
+                                "and WELL_ID='${header.well_id}' and PUMP_DEPTH is not null  group by WELL_ID)"))
                         .to("jdbc:oracle")
                         .split(body())
                         .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set pump_depth = '${body[PUMP_DEPTH]}'  " +
@@ -179,10 +183,11 @@ public class CamelJDBCConfiguration {
                         });
                 setSysControlBody(statusDailyBJ)
                         .setBody(simple("select distinct WELL_ID , PROD_DATE , PUMP_DIAMETER " +
-                                "from ZY_DXGW.PC_PRO_COM_DAILY " +
-                                "where  (WELL_ID,PROD_DATE) in" +
-                                " (SELECT  WELL_ID,max(PROD_DATE) PROD_DATE  FROM ZY_DXGW.PC_PRO_COM_DAILY" +
-                                "    WHERE PUMP_DIAMETER is not null and WELL_ID='${header.well_id}' group by WELL_ID)"))
+                                "                                from ZY_DXGW.PC_PRO_COM_DAILY " +
+                                "                                where (WELL_ID,PROD_DATE) in " +
+                                "                                 (SELECT WELL_ID, max(PROD_DATE) PROD_DATE  FROM ZY_DXGW.PC_PRO_COM_DAILY " +
+                                "                                  where   PROD_DATE > TO_DATE('2024-01-01','yyyy-MM-dd')  " +
+                                "and WELL_ID='${header.well_id}' and PUMP_DIAMETER is not null  group by WELL_ID)"))
                         .to("jdbc:oracle")
                         .split(body())
                         .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '${body[PUMP_DIAMETER]}'  " +
@@ -205,7 +210,7 @@ public class CamelJDBCConfiguration {
                         .split(body()).process(exchange -> {
                             Message in = exchange.getIn();
                             HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                            aRow.put("submergence_depth", null);
+                            aRow.put("submergence_depth", 0);
                             if (aRow.get("start_pump_liq_level") != null && aRow.get("pump_depth") != null) {
                                 double cmd = Double.parseDouble(aRow.get("pump_depth").toString()) - Double.parseDouble(aRow.get("start_pump_liq_level").toString()) / 10;
                                 BigDecimal bd = new BigDecimal(cmd);
@@ -217,7 +222,7 @@ public class CamelJDBCConfiguration {
                         .doTry()
                         .to("jdbc:centralbase")
                         .doCatch(Exception.class)
-                        .log("${header.date}" + " routeId:update-statusDaily-submergenceDepth ->  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();
 
 
@@ -227,15 +232,17 @@ public class CamelJDBCConfiguration {
                         .process(exchange -> {
                             Message in = exchange.getIn();
                             in.setHeader("date", getDate());
+                            in.setHeader("date", LocalDateTime.now().minusDays(1).toLocalDate().format(DateTimeFormatter.ofPattern("yyyy-MM-dd")));
                         });
                 setSysControlBody(volDaily)
                         .setBody(simple("select distinct  WELL_ID,PROD_DATE,PROD_TIME, LIQ_PROD_DAILY, OIL_PROD_DAILY,GAS_PROD_DAILY" +
-                                ",WATER_CUT, REMARKS from DBA01 where PROD_DATE  = to_date('${header.date}','yyyy-MM-dd')" +
+                                ",WATER_CUT, REMARKS from ZY_DXGW.PC_PRO_COM_DAILY where PROD_DATE  = to_date('${header.date}','yyyy-MM-dd')" +
                                 " and WELL_ID ='${header.well_id}'"))
                         .to("jdbc:oracle")
                         .split(body()).process(exchange -> {
                             Message in = exchange.getIn();
                             HashMap<String, Object> aRow = in.getBody(HashMap.class);
+
                             aRow.putIfAbsent("PROD_TIME", "0.0");
                             aRow.putIfAbsent("LIQ_PROD_DAILY", "0.0");
                             aRow.putIfAbsent("OIL_PROD_DAILY", "0.0");
@@ -276,6 +283,7 @@ public class CamelJDBCConfiguration {
                             if (!aRow.containsKey("YMD")) {
                                 aRow.put("YMD", 0.85);
                             }
+
                         })
                         .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,stim_content) " +
                                 "select '${body[WELL_ID]}','${body[PROD_DATE]}','${body[WATER_CUT]}','${body[LIQ_PROD_DAILY]}','${body[OIL_PROD_DAILY]}','${body[GAS_PROD_DAILY]}','${body[WATER_CUT]}','${body[REMARKS]}','${body[QYB]}','${body[RCSL]}','${body[SQB]}','${body[SMD]}','${body[YMD]}' ,'${body[SCCW]}'" +
@@ -300,7 +308,8 @@ public class CamelJDBCConfiguration {
                             in.setHeader("date", getDate());
                         });
                 setSysControlBody(volDailyLiqProdDaily)
-                        .setBody(simple("select distinct  WELL_ID,PROD_DATE,WATER_CUT, LIQ_PROD_DAILY,OIL_PROD_DAILY,GAS_PROD_DAILY,WATER_CUT, REMARKS from DBA01 where PROD_DATE  = to_date('${header.date}','yyyy-MM-dd')and WELL_ID ='${header.well_id}' and qyrq is not null "))
+                        .setBody(simple("select distinct  WELL_ID,PROD_DATE,WATER_CUT, LIQ_PROD_DAILY,OIL_PROD_DAILY,GAS_PROD_DAILY,WATER_CUT, REMARKS" +
+                                " from ZY_DXGW.PC_PRO_COM_DAILY where PROD_DATE  = to_date('${header.date}','yyyy-MM-dd')and WELL_ID ='${header.well_id}'"))
                         .to("jdbc:oracle")
                         .split(body()).process(exchange -> {
                             Message in = exchange.getIn();

src/main/resources/application.yml

@@ -131,7 +131,7 @@ gct:
     one:
       topic: diagnose-msg
       tags: v1
-      log-file-path: /home/lloyd/Desktop/log/task.json
+      log-file-path: C:\Users\Administrator\Desktop\package\a.json
       producer:
         group: diagnose
         access-key: diagnose-msg-v1
@@ -144,7 +144,7 @@ gct:
     two:
       topic: warn-msg
       tags: v1
-      log-file-path: /home/lloyd/Desktop/log/task.json
+      log-file-path: C:\Users\Administrator\Desktop\package\a.json
       producer:
         group: warn
         access-key: warn-msg-v1