4.19

pom.xml

@@ -57,6 +57,12 @@
     </distributionManagement>
 
     <dependencies>
+        <!--MQ-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+
         <!-- SpringBoot Web-->
         <dependency>
             <groupId>org.springframework.boot</groupId>

src/main/java/com/muyu/loadCenter/aliyun/service/AliYunEcsService.java

@@ -6,7 +6,7 @@ import com.aliyun.tea.TeaException;
 import com.aliyun.teautil.models.RuntimeOptions;
 import com.muyu.loadCenter.config.AliConfig;
 
-import com.muyu.loadCenter.aliyun.utils.UserUtil;
+import com.muyu.loadCenter.utils.UserUtil;
 import com.muyu.loadCenter.domain.EcsInstanceInfo;
 import com.muyu.loadCenter.config.InstanceConfig;
 import lombok.extern.log4j.Log4j2;

src/main/java/com/muyu/loadCenter/config/RabbitMqConfig.java

@@ -0,0 +1,86 @@
+package com.muyu.loadCenter.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+
+
+@Configuration
+public class RabbitMqConfig implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {
+    public static final Logger logger= LoggerFactory.getLogger(RabbitMqConfig.class);
+
+    private RabbitTemplate rabbitTemplate;
+
+
+    @Bean
+    public MessageConverter messageConverter(){
+        return new Jackson2JsonMessageConverter();
+    }
+
+    public static final String QUEUE="queue";
+    public static final String EXCHANGE="exchange";
+    public static final String KEY="Key";
+
+
+    @Primary
+    @Bean
+    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
+
+        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+
+        this.rabbitTemplate=rabbitTemplate;
+
+        rabbitTemplate.setMessageConverter(messageConverter());
+        rabbitTemplate();
+        return rabbitTemplate;
+    }
+
+    public void rabbitTemplate(){
+        rabbitTemplate.setConfirmCallback(this);
+        rabbitTemplate.setReturnsCallback(this);
+    }
+
+
+    @Bean
+    public Queue queue(){
+        return new Queue(QUEUE,true);
+    }
+
+
+    @Bean
+    public DirectExchange directExchange(){
+        return new DirectExchange(EXCHANGE);
+    }
+
+
+
+    @Bean
+    public Binding binding(){
+        return BindingBuilder.bind(queue()).to(directExchange()).with(KEY);
+    }
+
+
+
+    @Override
+    public void confirm(CorrelationData correlationData, boolean ack, String s) {
+
+        if (ack){
+            logger.info("{}消息到达交换机",correlationData.getId());
+        }else{
+            logger.error("{}消息丢失",correlationData.getId());
+        }
+    }
+
+    @Override
+    public void returnedMessage(ReturnedMessage returnedMessage) {
+        logger.error("{}消息未到达队列",returnedMessage.getMessage().getMessageProperties().getMessageId());
+    }
+}

src/main/java/com/muyu/loadCenter/controller/GatewayController.java

@@ -31,6 +31,8 @@ public class GatewayController {
     public Result<String> loadNode(){
         return Result.success(gatewayLoadService.loadNode());
     }
+
+
     /**
      * 定时任务，每30秒扫描一次服务器集群的负载情况。
      * 如果任一服务器的车辆连接数达到或超过阈值，则触发扩容；

src/main/java/com/muyu/loadCenter/mq/Consumer.java

@@ -0,0 +1,99 @@
+package com.muyu.loadCenter.mq;
+
+import com.muyu.loadCenter.aliyun.service.AliYunEcsService;
+import com.muyu.loadCenter.gateway.model.GatewayNodeInfo;
+import com.rabbitmq.client.Channel;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitHandler;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/**
+ * 消费者类，用于处理来自RabbitMQ的消息。
+ * 使用Spring AMQP框架监听指定队列中的消息。
+ */
+@RabbitListener(queues = "queue")
+@Component
+@Log4j2
+public class Consumer {
+
+    // 注入Spring的StringRedisTemplate以操作Redis
+    @Autowired
+    private StringRedisTemplate redisTemplate;
+
+    @Autowired
+    private AliYunEcsService aliYunEcsService;
+
+    /**
+     * 处理接收到的消息，具体逻辑包括：
+     * 1. 判断消息是否已被处理过，未处理则进行处理并标记为已处理；
+     * 2. 若消息已处理过，则认为是重复消费，拒绝处理。
+     * 3. 对于异常情况，根据重试次数决定是重试、拒绝还是退回消息。
+     *
+     * @param gatewayNodeInfo 消息中包含的网关节点信息。
+     * @param channel RabbitMQ的通道，用于确认消息处理情况。
+     * @param message 接收到的消息对象。
+     * @throws IOException 与RabbitMQ通信时可能抛出的异常。
+     */
+    @RabbitHandler
+    public void Handler(GatewayNodeInfo gatewayNodeInfo, Channel channel, Message message) throws IOException {
+        // 消息ID和传递标签
+        String messageId = message.getMessageProperties().getMessageId();
+
+        long deliveryTag = message.getMessageProperties().getDeliveryTag();
+
+        // 检查消息是否首次处理
+        if (!redisTemplate.hasKey("mi"+messageId)){
+            // 首次处理，存储传递标签
+            redisTemplate.opsForValue().set("mi"+messageId,deliveryTag+"");
+        }
+
+
+        try {
+            // 检查是否重复消费
+            if (!redisTemplate.hasKey("cf"+messageId)){
+                // 确认消费，并在Redis中做标记
+                log.info("确认消费");
+                log.info(gatewayNodeInfo);
+
+
+
+
+
+
+
+//                aliYunEcsService.releaseECS(gatewayNodeInfo.getNodeId()); // 释放ECS实例
+
+                redisTemplate.opsForValue().set("cf"+messageId,messageId);
+
+
+
+
+                channel.basicAck(deliveryTag,false);
+            }else {
+                // 重复消费，拒绝处理
+                log.info("重复消费");
+                channel.basicReject(deliveryTag,false);
+            }
+        } catch (Exception exception) {
+            // 异常处理逻辑，根据重试次数决定消息处理方式
+            String s = redisTemplate.opsForValue().get("mi" + messageId);
+            long old = Long.parseLong(s);
+
+            if (deliveryTag==(old+2)){
+                // 重试三次后，仍然失败，退回消息
+                log.info("已重试三次,实在消费不了");
+                channel.basicNack(deliveryTag,false,false);
+            }else {
+                // 尝试再次重试
+                log.info("继续重试");
+                channel.basicNack(deliveryTag,false,true);
+            }
+        }
+    }
+}

src/main/java/com/muyu/loadCenter/service/impl/GatewayLoadServiceImpl.java

@@ -3,6 +3,7 @@ package com.muyu.loadCenter.service.impl;
 import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
 import com.muyu.loadCenter.aliyun.service.AliYunEcsService;
+import com.muyu.loadCenter.config.RabbitMqConfig;
 import com.muyu.loadCenter.domain.EcsInstanceInfo;
 import com.muyu.loadCenter.domain.WorkGatewayNode;
 import com.muyu.loadCenter.gateway.cache.*;
@@ -14,12 +15,15 @@ import lombok.extern.log4j.Log4j2;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 
 /**
  * @ProjectName: load_center
@@ -53,7 +57,7 @@ public class GatewayLoadServiceImpl implements GatewayLoadService {
     private final GatewayNodeCache gatewayNodeCache;
 
     @Autowired
-    private RedisService redisService;
+    private RabbitTemplate rabbitTemplate;
 
 
 
@@ -125,7 +129,6 @@ public class GatewayLoadServiceImpl implements GatewayLoadService {
         gatewayLoadNodeCache.put(loadNodeList);
 
         Long seriesLoad = gatewayLoadSeriesCache.incrementAndGet();
-        System.out.println("seriesLoad:" + seriesLoad);
 
         Long seriesLoadIndex = seriesLoad % nodeLength;
 
@@ -178,7 +181,7 @@ public class GatewayLoadServiceImpl implements GatewayLoadService {
                 gatewayZSetNodeCache.put(ip,connectSize);
 
                 // 根据连接数判断是否需要进行扩容或缩容
-                if (connectSize >= 8) {
+                if (connectSize >= 5) {
                     aa++;
                     // 当满足扩容条件时，记录日志并执行扩容操作
                     if (aa == ipCacheSet.size()) {
@@ -199,10 +202,8 @@ public class GatewayLoadServiceImpl implements GatewayLoadService {
                         gatewayNodeInfo.setPublicIdAddress(ecsInstanceInfo.getPublicIpAddress());
                         gatewayNodeInfo.setPrivateIdAddress(ecsInstanceInfo.getPrivateIpAddress());
 
-
                         gatewayNodeCache.put(ecsInstanceInfo.getPublicIpAddress(), gatewayNodeInfo);
 
-
                         //这里模拟（也可以在别的类里完成） ECS创建成功后，服务器发送一条消息服务器正常启动，mq可以正常使用，存入redis
                         gatewayZSetNodeCache.put(ecsInstanceInfo.getPublicIpAddress(), 0);
                         log.info("实例id和公网ip存入redis");
@@ -219,18 +220,26 @@ public class GatewayLoadServiceImpl implements GatewayLoadService {
 
                     //缩容逻辑：删除连接数过低的服务器实例,出现连续两个节点连接数低于5的，则执行缩容操作
                     if (bb >= 2){
+
+
                         log.info("释放实例"+ip);
 
+
                         gatewayZSetNodeCache.remove(ip);
 
+
+
                         GatewayNodeInfo gatewayNodeInfo = gatewayNodeCache.get(ip);
 
+
                        //这里再来个异步操作 将 GatewayNodeInfo 对象传走,进行异步操作
+                        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE,RabbitMqConfig.KEY,gatewayNodeInfo, message -> {
+                            message.getMessageProperties().setMessageId(gatewayNodeInfo.getNodeId());
+                            return message;
+                        },new CorrelationData(UUID.randomUUID().toString()));
+
 
-//                        1.分批次解除连接的车辆
-//                        2.删除各个相关的redis缓存
 
-//                        aliYunEcsService.releaseECS(gatewayNodeInfo.getNodeId()); // 释放ECS实例
 
                         bb = 0;
 
@@ -240,25 +249,6 @@ public class GatewayLoadServiceImpl implements GatewayLoadService {
                     bb = 0;
                 }
 
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
             } catch (Exception e) {
                 e.printStackTrace();
             }

src/main/java/com/muyu/loadCenter/utils/UserUtil.java

@@ -1,7 +1,7 @@
-package com.muyu.loadCenter.aliyun.utils;
+package com.muyu.loadCenter.utils;
 
 
-import com.muyu.loadCenter.aliyun.utils.uuid.IdUtils;
+import com.muyu.loadCenter.utils.uuid.IdUtils;
 import com.muyu.loadCenter.redis.service.RedisService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

src/main/java/com/muyu/loadCenter/utils/uuid/IdUtils.java

@@ -1,4 +1,4 @@
-package com.muyu.loadCenter.aliyun.utils.uuid;
+package com.muyu.loadCenter.utils.uuid;
 
 /**
  * ID生成器工具类

src/main/java/com/muyu/loadCenter/utils/uuid/UUID.java

@@ -1,4 +1,4 @@
-package com.muyu.loadCenter.aliyun.utils.uuid;
+package com.muyu.loadCenter.utils.uuid;
 
 
 import java.security.MessageDigest;

src/main/resources/application.yml

@@ -10,6 +10,19 @@ Spring:
     port: 6379
 
 
+  rabbitmq:
+    username: guest
+    password: guest
+    virtualHost: /
+    port: 5672
+    host: 121.89.221.195
+    listener:
+      simple:
+        prefetch: 1 # 每次只能获取一条，处理完成才能获取下一条
+    publisher-confirm-type: correlated   #确认消息已发送到交换机(Exchange)
+    publisher-returns: true  #确认消息已发送到队列(Queue)
+
+
 config:
   ali:
     access-key-id: LTAI5tANGefs2gi8nsu4AoSZ

src/test/java/LoadTest.java

@@ -22,8 +22,26 @@ public class LoadTest {
 
     @Test
     public void load() {
-        GatewayNodeInfo gatewayNodeInfo = redisService.getCacheObject("gateway:node:info:39.101.193.188");
-        System.out.println(gatewayNodeInfo);
+
+
+
+      redisService.setCacheZSet("gateway:zSet:count","39.103.130.89",0);
+      redisService.setCacheZSet("gateway:zSet:count","121.89.221.195",0);
+
+//        GatewayNodeInfo gatewayNodeInfo = new GatewayNodeInfo();
+//        gatewayNodeInfo.setPublicIdAddress("39.103.130.89");
+//        gatewayNodeInfo.setPrivateIdAddress("39.103.130.89");
+//        gatewayNodeInfo.setNodeId("i-8vbctj4u8ohgaqt9wwre");
+//
+//        redisService.setCacheObject("gateway:node:info:39.103.130.89",gatewayNodeInfo);
+
+
+//        gatewayNodeInfo.setPublicIdAddress("121.89.221.195");
+//        gatewayNodeInfo.setPrivateIdAddress("121.89.221.195");
+//        gatewayNodeInfo.setNodeId("i-8vbemtl328dk8er33ylv");
+//
+//        redisService.setCacheObject("gateway:node:info:121.89.221.195",gatewayNodeInfo);
+
 
 
     }