向诊断队列发送消息

pom.xml

@@ -130,6 +130,10 @@
             <artifactId>camel-servlet</artifactId>
         </dependency>
         <dependency>
+            <groupId>org.apache.camel.springboot</groupId>
+            <artifactId>camel-quartz-starter</artifactId>
+        </dependency>
+        <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>

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

@@ -0,0 +1,63 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
+import java.util.logging.FileHandler;
+import java.util.logging.Formatter;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+
+/**
+ * class name: DiagnoseMsgSendFailCallBack
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/7/05 下午7:08
+ */
+@Slf4j
+public class DefaultMsgSendFailCallBack {
+
+    public static void accept(MessageBody msg, String logFilePath) {
+        String msgJson = msg.toJsonBody();
+        String fileName = logFilePath + "-" + LocalDate.now().format(DateTimeFormatter.ISO_DATE);
+        File file = new File(fileName);
+        if (!file.getParentFile().exists())
+            if (!file.getParentFile().mkdirs()) {
+                log.error("create log file failed,method:mkdirs()");
+                return;
+            }
+        if (!file.exists()) {
+            try {
+                if (!file.createNewFile()) {
+                    log.error("create log file failed,method:createNewFile()");
+                    return;
+                }
+            } catch (IOException ioException) {
+                log.error("create log file failed, cause by:{}", ioException.getMessage());
+                return;
+            }
+        }
+
+        try {
+            FileHandler fileHandler = new FileHandler(fileName, true);
+            fileHandler.setFormatter(new Formatter() {
+                @Override
+                public String format(LogRecord logRecord) {
+                    return logRecord.getMessage();
+                }
+            });
+            Logger logger = Logger.getLogger(DefaultMsgSendFailCallBack.class.getTypeName());
+            logger.setUseParentHandlers(false);
+            logger.addHandler(fileHandler);
+            logger.info(msgJson + "\n");
+        } catch (IOException ioException) {
+            log.error("run logger handler failed, cause by:{}", ioException.getMessage());
+        }
+
+
+    }
+}

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

@@ -0,0 +1,23 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.springframework.stereotype.Component;
+
+/**
+ * class name: DiagnoseMsgSendSuccessCallBack
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午7:23
+ */
+@Slf4j
+@Component("defaultMsgSendSuccessCallBack")
+public class DefaultMsgSendSuccessCallBack implements SendSuccessCallBack {
+    private static int count = 0;
+    @Override
+    public void accept(SendResult sendResult) {
+        System.out.println("send = "+count++);
+        log.info("send msg success,msg:{}", sendResult.getMsgId());
+    }
+}

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

@@ -0,0 +1,41 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+
+/**
+ * class name: DiagnoseDelayedMessageProducer
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午7:41
+ */
+public class DiagnoseDelayedMessageProducer implements MessageProducer {
+    DefaultMQProducer producer;
+
+    @Override
+    public void destroy() {
+        producer.shutdown();
+    }
+
+    @Override
+    public void init() {
+        try {
+            producer.start();
+        } catch (MQClientException e) {
+            System.out.println("Error");
+        }
+    }
+
+    //@Scheduled(zone = "****")
+    public void task() {
+        /**
+         * TODO:
+         * 完善延迟重发策略以及统一补偿策略
+         * 间隔一段时间将logfile中的统一生产到mq中
+         * 默认7天2G（循环擦除1小时）限制，理论上本地空间无限保留时间无限
+         */
+        //Executors.newScheduledThreadPool(1).schedule()
+        //Quartz
+    }
+}

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

@@ -0,0 +1,60 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import com.alibaba.fastjson.JSONObject;
+import com.gct.tools.etlcamelhuge.entity.DiagnoseMsg;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.message.Message;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * class name: DiagnoseMessageProducer
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:44
+ */
+@Slf4j
+public class DiagnoseMessageProducer implements MessageProducer {
+    private final DefaultMQProducer producer;
+    private final GCTRocketMQProperties properties;
+
+    private final SendFailCallBack failCallBack;
+    private final SendSuccessCallBack successCallBack;
+
+    public DiagnoseMessageProducer(DefaultMQProducer producer, GCTRocketMQProperties properties, SendFailCallBack failCallBack, SendSuccessCallBack successCallBack) {
+        this.producer = producer;
+        this.properties = properties;
+        this.failCallBack = failCallBack;
+        this.successCallBack = successCallBack;
+    }
+
+
+    @Override
+    public void init() {
+        try {
+            producer.start();
+        } catch (MQClientException e) {
+            log.error("producer can not start ,cause by:{}", e.getErrorMessage());
+        }
+    }
+
+    @Override
+    public void send(MessageBody msgBody) {
+        MessageProducer.super.send(msgBody);
+        if (!(msgBody instanceof DiagnoseMsg)) {
+            log.error("Not supported type of msg body:{},and only supported 'DiagnoseMsg'", msgBody.getClass().getTypeName());
+            return;
+        }
+        DiagnoseMsg actualMsg = (DiagnoseMsg) msgBody;
+        Message msg = new Message(properties.getTopic(), properties.getTags(), actualMsg.getWellId(), JSONObject.toJSONString(actualMsg).getBytes(StandardCharsets.UTF_8));
+        sendDefault(producer, msg, (MessageBody) actualMsg, failCallBack, successCallBack);
+    }
+
+    @Override
+    public void destroy() {
+        producer.shutdown();
+    }
+}

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

@@ -0,0 +1,18 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * class name: GCTMutiableMQProperties
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/7/2 下午2:37
+ */
+@ConfigurationProperties(prefix = "gct.rocketmq")
+@Data
+public class GCTMutableMQProperties {
+    private GCTRocketMQProperties one;
+    private GCTRocketMQProperties two;
+}

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

@@ -0,0 +1,74 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.spring.support.RocketMQUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * class name: GCTRocketMQConfiguration
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:30
+ */
+@Configuration
+@EnableConfigurationProperties({GCTMutableMQProperties.class})
+public class GCTRocketMQConfiguration {
+
+    public static DefaultMQProducer buildDefaultMQProducer(GCTRocketMQProperties properties) {
+        GCTRocketMQProperties.Producer producerConfig = properties.getProducer();
+        String nameServer = properties.getNameServer();
+        String groupName = producerConfig.getGroup();
+        Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
+        Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
+        String accessChannel = properties.getAccessChannel();
+        String ak = properties.getProducer().getAccessKey();
+        String sk = properties.getProducer().getSecretKey();
+        boolean isEnableMsgTrace = properties.getProducer().isEnableMsgTrace();
+        String customizedTraceTopic = properties.getProducer().getCustomizedTraceTopic();
+        DefaultMQProducer producer = RocketMQUtil.createDefaultMQProducer(groupName, ak, sk, isEnableMsgTrace, customizedTraceTopic);
+        producer.setNamesrvAddr(nameServer);
+        if (!StringUtils.isEmpty(accessChannel)) {
+            producer.setAccessChannel(AccessChannel.valueOf(accessChannel));
+        }
+
+        producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
+        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
+        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
+        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
+        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMessageBodyThreshold());
+        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
+        return producer;
+    }
+
+    @Bean(name = "diagnoseMessageProducer", destroyMethod = "destroy", initMethod = "init")
+    public MessageProducer diagnoseMessageProducer(GCTMutableMQProperties properties,
+                                                   @Qualifier("defaultMsgSendSuccessCallBack") SendSuccessCallBack successCallBack) {
+        GCTRocketMQProperties one = properties.getOne();
+        return new DiagnoseMessageProducer(buildDefaultMQProducer(one), one, (e, msg) -> {
+            Logger log = LoggerFactory.getLogger(DiagnoseMessageProducer.class);
+            log.error("send diagnose message failed  cause by:{},stack:{}", e.getMessage(), e.getStackTrace());
+            DefaultMsgSendFailCallBack.accept(msg, one.getLogFilePath());
+        }, successCallBack);
+    }
+
+    @Bean(name = "warnMessageProducer", destroyMethod = "destroy", initMethod = "init")
+    public MessageProducer warnMessageProducer(GCTMutableMQProperties properties,
+                                               @Qualifier("defaultMsgSendSuccessCallBack") SendSuccessCallBack successCallBack) {
+        GCTRocketMQProperties two = properties.getTwo();
+        return new WarnMessageProducer(buildDefaultMQProducer(two), two, (e, msg) -> {
+            Logger log = LoggerFactory.getLogger(WarnMessageProducer.class);
+            log.error("send warn message failed  cause by:{},stack:{}", e.getMessage(), e.getStackTrace());
+            DefaultMsgSendFailCallBack.accept(msg, two.getLogFilePath());
+        }, successCallBack);
+    }
+
+}

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

@@ -0,0 +1,22 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+
+/**
+ * class name: GCTRocketMQProperties
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:26
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+public class GCTRocketMQProperties extends RocketMQProperties {
+
+    private String topic;
+    private String logFilePath;
+    private String tags;
+
+}

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

@@ -0,0 +1,12 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+/**
+ * interface name: MessageBody
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午7:01
+ */
+public interface MessageBody {
+    String toJsonBody();
+}

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

@@ -0,0 +1,51 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendCallback;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+
+/**
+ * class name: MessageProducer
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:41
+ */
+
+public interface MessageProducer {
+    void init();
+
+    default void send(MessageBody msgBody) {
+        LoggerFactory.getLogger(MessageProducer.class).info("sending msg :==>{}", msgBody.toJsonBody());
+    }
+
+    void destroy() throws Exception;
+
+    default void sendDefault(DefaultMQProducer producer, Message msg, MessageBody actualMsg, SendFailCallBack failCallBack, SendSuccessCallBack successCallBack) {
+        Logger log = LoggerFactory.getLogger(producer.getProducerGroup() + producer.getClass().getTypeName());
+        try {
+            producer.send(msg, new SendCallback() {
+                @Override
+                public void onSuccess(SendResult sendResult) {
+                    if (Objects.nonNull(successCallBack)) successCallBack.accept(sendResult);
+                }
+
+                @Override
+                public void onException(Throwable throwable) {
+                    if (Objects.nonNull(failCallBack)) failCallBack.accept(throwable, actualMsg);
+                }
+            });
+        } catch (MQClientException | RemotingException | InterruptedException e) {
+            log.error("Send msg failed,by cause:{},stack:{}", e.getMessage(), e.getStackTrace());
+        }
+    }
+
+
+}

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

@@ -0,0 +1,13 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+/**
+ * interface name: SendFailCallBack
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:58
+ */
+@FunctionalInterface
+public interface SendFailCallBack {
+    void accept(Throwable e, MessageBody msg);
+}

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

@@ -0,0 +1,15 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import org.apache.rocketmq.client.producer.SendResult;
+
+/**
+ * interface name: SendSuccessCallBack
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午7:03
+ */
+@FunctionalInterface
+public interface SendSuccessCallBack {
+    void accept(SendResult sendResult);
+}

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

@@ -0,0 +1,57 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.common.message.Message;
+
+import java.nio.charset.StandardCharsets;
+
+/**
+ * class name: DiagnoseMessageProducer
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:44
+ */
+@Slf4j
+public class WarnMessageProducer implements MessageProducer {
+    private final DefaultMQProducer producer;
+    private final GCTRocketMQProperties properties;
+
+    private final SendFailCallBack failCallBack;
+    private final SendSuccessCallBack successCallBack;
+
+    public WarnMessageProducer(DefaultMQProducer producer, GCTRocketMQProperties properties, SendFailCallBack failCallBack, SendSuccessCallBack successCallBack) {
+        this.producer = producer;
+        this.properties = properties;
+        this.failCallBack = failCallBack;
+        this.successCallBack = successCallBack;
+    }
+
+    @Override
+    public void init() {
+        try {
+            producer.start();
+        } catch (MQClientException e) {
+            log.error("producer can not start ,cause by:{}", e.getErrorMessage());
+        }
+    }
+
+    @Override
+    public void send(MessageBody msgBody) {
+        MessageProducer.super.send(msgBody);
+        if (!(msgBody instanceof WarnMsg)) {
+            log.error("Not supported type of msg body:{},and only supported 'WarnMsg'", msgBody.getClass().getTypeName());
+            return;
+        }
+        WarnMsg actualMsg = (WarnMsg) msgBody;
+        Message msg = new Message(properties.getTopic(), properties.getTags(), actualMsg.getWellId(), msgBody.toJsonBody().getBytes(StandardCharsets.UTF_8));
+        sendDefault(producer, msg, actualMsg, failCallBack, successCallBack);
+    }
+
+    @Override
+    public void destroy() {
+        producer.shutdown();
+    }
+}

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

@@ -0,0 +1,39 @@
+package com.gct.tools.etlcamelhuge.MQ;
+
+import com.alibaba.fastjson.JSONObject;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+/**
+ * class name: MQDiagnoseMsg
+ * <p>
+ * rocketMQ exchange message entity:
+ * for diagnose process
+ * <p>
+ * Notice:
+ * 1.the message key:well_id
+ * 2.the formatter of sgt eg:0,123,1,253...
+ * 3.use the secret "DiagnoseMsg_v1.0" to send message
+ * 4.in order to send message sequence orderly,please do that :
+ * sendOneWayOrderly(./././)
+ * </p>
+ *
+ * @author lloyd
+ * @version 1.0
+ * @since 2021/6/24 下午6:48
+ */
+@Data
+@AllArgsConstructor
+public class WarnMsg implements MessageBody {
+    private String wellId;
+    private String context;
+    private String wnDate;
+    private String resultId;
+    private String category;
+    private String rank;
+
+    @Override
+    public String toJsonBody() {
+        return JSONObject.toJSONString(this);
+    }
+}

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

@@ -0,0 +1,29 @@
+package com.gct.tools.etlcamelhuge.entity;
+
+import com.alibaba.fastjson.JSONObject;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.gct.tools.etlcamelhuge.MQ.MessageBody;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class DiagnoseMsg implements MessageBody {
+
+    private String wellId;
+    private String wellName;
+    private String orgId;
+
+    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss" , timezone = "GMT+8")
+    private String prodDate;
+    private String sgt;
+    private String rcvDate;
+    private double s;
+    private double n;
+
+
+    @Override
+    public String toJsonBody() {
+        return JSONObject.toJSONString(this);
+    }
+}

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

@@ -1,14 +0,0 @@
-package com.gct.tools.etlcamelhuge.entity;
-
-import lombok.Data;
-
-@Data
-public class MQDiagnoseMsg {
-
-    private String well_id ;
-    private String well_name;
-    private String org_id;
-    private String pord_date;//创建时间
-    private String sgt;
-    private String rcv_date;//采集时间
-}

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

@@ -1,26 +1,21 @@
 package com.gct.tools.etlcamelhuge.routeconfig;
 
-import com.alibaba.fastjson.JSONObject;
-import com.gct.tools.etlcamelhuge.entity.MQDiagnoseMsg;
+import com.gct.tools.etlcamelhuge.MQ.MessageBody;
+import com.gct.tools.etlcamelhuge.MQ.MessageProducer;
+import com.gct.tools.etlcamelhuge.entity.DiagnoseMsg;
 import org.apache.camel.*;
 import org.apache.camel.builder.RouteBuilder;
-import org.apache.rocketmq.client.producer.DefaultMQProducer;
 //import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.slf4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.http.HttpEntity;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.MediaType;
-import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator;
 
+import javax.annotation.Resource;
 import java.math.BigDecimal;
-import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
+import java.time.LocalDateTime;
 import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * class name: CamelJDBCConfiguration
@@ -271,6 +266,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
         }
         return Arrays.stream(doubles).max().getAsDouble();
     }
+    @Resource(name = "diagnoseMessageProducer")
+    private MessageProducer producer;
     @Bean
     public RouteBuilder routeBuilderWithOracle1() {
         SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
@@ -285,9 +282,9 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
             //全部执行完成的大概时间在30-40分钟
             @Override
             public void configure() throws Exception {
-                //0 0 18 * * ?   每天下午6点执行一次
+                //24小时执行一次
                 //单个执行时间30s左右,在之前有数据的情况下
-                from("timer:mytimer1?period=99999999999")
+                from("timer:mytimer1?period=604800000")
                         .routeId("oracle-1")
                         .setHeader("date", constant(date1))
                         .setBody(simple("select  distinct jh,cydmc,zyq,zk,qyrq,sccw,qk,bz from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
@@ -372,7 +369,7 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         .log("insert")
                         .end();
                 //单独执行时间10s
-                from("timer:mytimer2?period=99999999999")
+                from("timer:mytimer2?period=3600000")
                         .routeId("oracle-2")
                         .setHeader("date", constant(date1))
                         .setBody(simple("select  distinct jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
@@ -396,7 +393,7 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                 //将查询到的DYM数据更新到cb_pc_pro_wellbore_status_daily中
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行时间是4m15s 317条数据
-                from("timer:mytimer5?period=99999999999")
+                from("timer:mytimer5?period=3600000")
                         .routeId("oracle-5")
                         .setHeader("date", constant(date1 + " 00:00:00"))
                         //三个月之内dym不为空的数据
@@ -409,7 +406,7 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         .log("insert !!!")
                         .end();
                 //单独执行时间30s
-                from("timer:mytimer3?period=99999999999")
+                from("timer:mytimer3?period=3600000")
                         .routeId("oracle-3")
                         .setHeader("date", constant(date1))
                         .setBody(simple("select distinct  jh,rq,scsj, rcyl1,rcyl,rcql,hs, bz from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
@@ -432,7 +429,7 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
 
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行一次30s
-                from("timer:mytimer4?period=99999999999")
+                from("timer:mytimer4?period=3600000")
                         .routeId("oracle-4")
                         .setHeader("date", constant(date1))
                         .setBody(simple("select distinct jh,rq,dym,jy,ly,bj,bs,bx,zs,cc,cs,blx,dl from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null  "))
@@ -460,7 +457,7 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                 //从天安哪里获取的数据
                 //0 0 */1 * * ? 每1个小时执行一次
                 //单独执行一小时的数据30s
-                from("timer:mytimer7?period=99999999999")
+               /* from("timer:mytimer7?period=3600000")
                         .routeId("jdbc-gtsj-?")
                         .setBody(simple("select max(prod_date) from centralbase.cb_temp_well_mech_runtime "))
                         .to("jdbc:centralbase")
@@ -501,9 +498,29 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         .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) " +
                                 "values ('${body[well_name]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}')"))
                         .to("jdbc:centralbase")
-                        .log("insert!!!").end();
+                        .log("insert!!!").end();*/
+
+                from("timer:mytimer1?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")
+                        .split(body()).log("log{body}")
+                        .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());
+                            System.out.println("strokeFrequency"+strokeFrequency);
+                            String sgt = aRow.get("sgt").toString();
+                            DiagnoseMsg diagnoseMsg = new DiagnoseMsg(wellId, wellName, orgId, prodDate, sgt, LocalDateTime.now().toString(), strokeLength, strokeFrequency);
+                            producer.send((MessageBody) diagnoseMsg);
+                        }).log("send success").end();
 //-------------------------------------------------------------------------------------------------------------------------------------------------------------------
-           /*     from("timer:mytimer2?period=999999999")
+           /*     from("timer:mytimer2?period=3600000")
                         .routeId("oracle-2")
                         .setBody(simple("select distinct station_id,station_name from centralbase.cb_cd_well_source"))
                         .to("jdbc:centralbase")

src/main/resources/application.yml

@@ -124,10 +124,31 @@ rocketmq:
     sendMessageTimeout: 300000
     access-key: 123456
     secret-key: 123456
-logging:
-  file:
-    path: /home/gxt/IdeaProjects/etl-camel-huge/logs
-    max-history: 30
-    max-size: 10
-    name: log
-
+gct:
+  rocketmq:
+    one:
+      topic: diagnose-msg
+      tags: v1
+      log-file-path: /home/lloyd/Desktop/log/task.json
+      producer:
+        group: diagnose
+        access-key: diagnose-msg-v1
+        secret-key: diagnose-msg-v1
+      name-server: 10.72.143.2:9876
+      consumer:
+        group: diagnose
+        access-key: diagnose-msg-v1
+        secret-key: diagnose-msg-v1
+    two:
+      topic: warn-msg
+      tags: v1
+      log-file-path: /home/lloyd/Desktop/log/task.json
+      producer:
+        group: warn
+        access-key: warn-msg-v1
+        secret-key: warn-msg-v1
+      name-server: 10.72.143.2:9876
+      consumer:
+        group: warn
+        access-key: warn-msg-v1
+        secret-key: warn-msg-v1