test 环境当前版本

修改 rabbitmq 配置

common/src/main/java/cn/reghao/autodop/common/amqp/RabbitProducer.java

@@ -2,9 +2,13 @@ package cn.reghao.autodop.common.amqp;
 
 import cn.reghao.autodop.common.utils.serializer.JsonConverter;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.stereotype.Component;
 
+import java.util.UUID;
+
 /**
  * MQ 消息生产者
  *
@@ -18,6 +22,23 @@ public class RabbitProducer {
 
     public RabbitProducer(RabbitTemplate rabbitTemplate) {
         this.rabbitTemplate = rabbitTemplate;
+        config();
+    }
+
+    private void config() {
+        // 开启可靠投递
+        rabbitTemplate.setConfirmCallback((data, ack, cause) -> {
+            if (!ack) {
+                log.error("消息发送失败 -> {}", cause + (data != null ? data.toString() : null));
+            } else {
+                log.info("消息发送成功 -> {}", cause + (data != null ? data.getId() : null));
+            }
+        });
+
+        rabbitTemplate.setMandatory(true);
+        rabbitTemplate.setReturnCallback(((message, replyCode, replyText, exchange, routingKey) ->
+                log.info("消息发送 ReturnCallback: {}, {}, {}, {}, {}",
+                        message, replyCode, replyText, exchange, routingKey)));
     }
 
     /**
@@ -27,9 +48,11 @@ public class RabbitProducer {
      */
     public void send(String exchange, String routeKey, MqMessage mqMessage) {
         String msg = JsonConverter.objectToJson(mqMessage);
-        rabbitTemplate.convertAndSend(exchange, routeKey, msg);
         // 默认的 exchange 是 amq.direct
-        //rabbitTemplate.convertAndSend(routeKey, msg);
+        rabbitTemplate.convertAndSend(exchange, routeKey, msg, message -> {
+            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
+            return message;
+        }, new CorrelationData(UUID.randomUUID().toString()));
     }
 
     /**
@@ -40,6 +63,7 @@ public class RabbitProducer {
      * @date 2021-02-23 下午3:57
      */
     public RpcResult callRemote(String exchange, String routeKey, MqMessage mqMessage) {
+        // TODO 处理 RPC 调用在超时后才成功的情况
         rabbitTemplate.setReplyTimeout(90_000);
         String msg = JsonConverter.objectToJson(mqMessage);
         String result = (String) rabbitTemplate.convertSendAndReceive(exchange, routeKey, msg);

dagent/src/main/java/cn/reghao/autodop/dagent/utils/amqp/DagentConsumerConfig.java

@@ -10,6 +10,9 @@ import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * dagent 队列消费者
  *
@@ -32,7 +35,11 @@ public class DagentConsumerConfig {
     @Bean
     public Queue dagentQueue() {
         String queueName = "dagent@" + Machine.machineId();
-        return new Queue(queueName, true);
+        Map<String, Object> args = new HashMap<>();
+        // 队列 TTL
+        args.put("x-message-ttl", 6000);
+        // 创建一个队列
+        return new Queue(queueName, true, false, false, args);
     }
 
     @Bean

dagent/src/main/java/cn/reghao/autodop/dagent/utils/log/LoggerConfig.java

@@ -43,7 +43,7 @@ public class LoggerConfig {
         amqpAppender.setPassword(rabbitProperties.getPassword());
         amqpAppender.setDeclareExchange(true);
         amqpAppender.setExchangeType("direct");
-        amqpAppender.setExchangeName("rabbit.log");
+        amqpAppender.setExchangeName("rabbit.dagent.log");
         amqpAppender.setRoutingKeyPattern("info");
         amqpAppender.setDurable(true);
         amqpAppender.setAutoDelete(false);

dagent/src/main/java/cn/reghao/autodop/dagent/utils/log/LoggingInitializer.java

@@ -44,7 +44,7 @@ public class LoggingInitializer implements ApplicationContextInitializer<Configu
         amqpAppender.setPassword("Dev@12345678");
         amqpAppender.setDeclareExchange(true);
         amqpAppender.setExchangeType("direct");
-        amqpAppender.setExchangeName("rabbit.log");
+        amqpAppender.setExchangeName("rabbit.dagent.log");
         amqpAppender.setRoutingKeyPattern("info");
         amqpAppender.setDurable(true);
         amqpAppender.setAutoDelete(false);

dagent/src/main/resources/application-dev.yml

@@ -1,7 +1,13 @@
 spring:
+#  rabbitmq:
+#    host: localhost
+#    port: 5672
+#    virtual-host: /
+#    username: dev
+#    password: Dev@12345678
   rabbitmq:
-    host: localhost
+    host: mq.srv.iquizoo.com
     port: 5672
-    virtual-host: /
-    username: dev
-    password: Dev@12345678
+    username: iquizoo-mq1
+    password: z2pT1PXR
+    virtual-host: /

dagent/src/main/resources/application-test.yml

@@ -1,14 +1,7 @@
-#spring:
-#  rabbitmq:
-#    host: mq.srv.iquizoo.com
-#    port: 5672
-#    username: iquizoo-mq1
-#    password: z2pT1PXR
-#    virtual-host: /
 spring:
   rabbitmq:
-    host: 192.168.0.111
+    host: mq.srv.iquizoo.com
     port: 5672
-    virtual-host: /
-    username: dev
-    password: Dev@12345678
+    username: iquizoo-mq1
+    password: z2pT1PXR
+    virtual-host: /

dmaster/src/main/java/cn/reghao/autodop/dmaster/utils/amqp/RabbitConsumer.java → dmaster/src/main/java/cn/reghao/autodop/dmaster/utils/amqp/DmasterConsumer.java

@@ -19,15 +19,17 @@ import org.springframework.stereotype.Component;
  */
 @Slf4j
 @Component
-public class RabbitConsumer {
+public class DmasterConsumer {
     private MachineDispatcher machineDispatcher;
 
-    public RabbitConsumer(MachineDispatcher machineDispatcher) {
+    public DmasterConsumer(MachineDispatcher machineDispatcher) {
         this.machineDispatcher = machineDispatcher;
     }
 
+    // 将 dmaster 队列通过 routeKey dmaster 绑定到 amq.direct 交换机
     @RabbitListener(bindings = @QueueBinding(
             value = @Queue(value = "dmaster", durable = "true"),
+            key = "dmaster",
             exchange = @Exchange(value = "amq.direct"))
     )
     public void dmasterQueueConsumer(@Payload String msg) {
@@ -55,8 +57,8 @@ public class RabbitConsumer {
     }
 
     @RabbitListener(bindings = @QueueBinding(
-            value = @Queue(value = "log", durable = "true"),
-            exchange = @Exchange(value = "rabbit.log"))
+            value = @Queue(value = "dagent.log", durable = "true"),
+            exchange = @Exchange(value = "rabbit.dagent.log"))
     )
     public void logQueueConsumer(@Payload String msg) {
         //log.info("日志消息 -> {}", msg);

dmaster/src/main/resources/application-dev.yml

@@ -6,15 +6,15 @@ spring:
   data:
     mongodb:
       uri: mongodb://localhost/reghao_devops_rdb
-  rabbitmq:
-    host: localhost
-    port: 5672
-    virtual-host: /
-    username: dev
-    password: Dev@12345678
 #  rabbitmq:
-#    host: mq.srv.iquizoo.com
+#    host: localhost
 #    port: 5672
-#    username: iquizoo-mq1
-#    password: z2pT1PXR
-#    virtual-host: /
+#    virtual-host: /
+#    username: dev
+#    password: Dev@12345678
+  rabbitmq:
+    host: mq.srv.iquizoo.com
+    port: 5672
+    username: iquizoo-mq1
+    password: z2pT1PXR
+    virtual-host: /

scripts/build.sh

@@ -15,7 +15,7 @@ proj_home="/home/reghao/code/aha/autodop"
 proj=`basename ${proj_home}`
 app_name=${proj}-${app}
 cd ${proj_home}
-mvn clean package -am -pl ${app} -Dmaven.test.skip=true -P${env}
+mvn clean package -am -Dmaven.test.skip=true -pl ${app} -P${env}
 
 # 存放应用的根目录
 jar_home="/home/reghao/dev/tmp/apps"

scripts/runner.sh

@@ -20,7 +20,15 @@ stop() {
 start() {
   nohup java -jar ${jar_file} > ${console_log} 2>&1 &
   pid=`jps | grep ${app_name} | awk '{print $1}'`
-  echo ${pid} > ${app_name}.pid
+  #echo ${pid} > ${app_name}.pid
+  if [ ! -z ${pid} ]
+  then
+    echo ${pid} > ${app_name}.pid
+    echo "${app_name} has been started with pid ${pid}..."
+  else
+    echo "${app_name} started failed, please check start log in log file console.log..."
+    exit 1
+  fi
 }
 
 case $1 in