mq send retry one time

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

@@ -20,6 +20,6 @@ public class DefaultMsgSendSuccessCallBack implements SendSuccessCallBack {
     @Override
     public void accept(SendResult sendResult) {
         //System.out.println("send success= "+count.incrementAndGet());
-        //log.info("send msg success,msg:{}", sendResult.getMsgId());
+        log.info("send msg success,msg:{}", sendResult.getMsgId());
     }
 }

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

@@ -50,7 +50,7 @@ public class DiagnoseMessageProducer implements MessageProducer {
         }
         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);
+        sendDefault(producer, msg, (MessageBody) actualMsg, failCallBack, successCallBack,0);
     }
 
     @Override

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

@@ -24,12 +24,12 @@ public interface MessageProducer {
     void init();
 
     default void send(MessageBody msgBody) {
-        LoggerFactory.getLogger(MessageProducer.class).info("sending msg :==>{}",((DiagnoseMsg)msgBody).getWellId()+" "+ ((DiagnoseMsg)msgBody).getProdDate()+" "+((DiagnoseMsg)msgBody).getSgt().substring(0,5));
+        LoggerFactory.getLogger(MessageProducer.class).info("sending msg :==>{}", ((DiagnoseMsg) msgBody).getWellId() + " " + ((DiagnoseMsg) msgBody).getProdDate() + " " + ((DiagnoseMsg) msgBody).getSgt().substring(0, 5));
     }
 
     void destroy() throws Exception;
 
-    default void sendDefault(DefaultMQProducer producer, Message msg, MessageBody actualMsg, SendFailCallBack failCallBack, SendSuccessCallBack successCallBack) {
+    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() {
@@ -40,7 +40,12 @@ public interface MessageProducer {
 
                 @Override
                 public void onException(Throwable throwable) {
-                    if (Objects.nonNull(failCallBack)) failCallBack.accept(throwable, actualMsg);
+                    if (retryCount < 2) {
+                        log.debug("retry send msg ,last time failed cause by {}",throwable.getMessage());
+                        sendDefault(producer, msg, actualMsg, failCallBack, successCallBack, retryCount + 1);
+                    } else {
+                        if (Objects.nonNull(failCallBack)) failCallBack.accept(throwable, actualMsg);
+                    }
                 }
             });
         } catch (MQClientException | RemotingException | InterruptedException e) {

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

@@ -47,7 +47,7 @@ public class WarnMessageProducer implements MessageProducer {
         }
         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);
+        sendDefault(producer, msg, actualMsg, failCallBack, successCallBack,0);
     }
 
     @Override