feat():before testRunner

version-2():解决迭代器报空异常

.idea/inspectionProfiles/Project_Default.xml

@@ -0,0 +1,17 @@
+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="AliAccessStaticViaInstance" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliArrayNamingShouldHaveBracket" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliControlFlowStatementWithoutBraces" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliDeprecation" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliEqualsAvoidNull" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliLongLiteralsEndingWithLowercaseL" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliMissingOverrideAnnotation" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AliWrapperTypeEquality" enabled="true" level="WARNING" enabled_by_default="true" />
+    <inspection_tool class="AutoCloseableResource" enabled="true" level="WARNING" enabled_by_default="true">
+      <option name="METHOD_MATCHER_CONFIG" value="java.util.Formatter,format,java.io.Writer,append,com.google.common.base.Preconditions,checkNotNull,org.hibernate.Session,close,java.io.PrintWriter,printf,java.io.PrintStream,printf,com.mobai.mq.rabbitmq.cofig.MqttFactory,buildOptions" />
+    </inspection_tool>
+    <inspection_tool class="MapOrSetKeyShouldOverrideHashCodeEquals" enabled="true" level="WARNING" enabled_by_default="true" />
+  </profile>
+</component>

pom.xml

@@ -55,6 +55,12 @@
             <version>2.0.46</version>
         </dependency>
 
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.83</version>
+        </dependency>
+
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
@@ -73,6 +79,40 @@
             <version>5.1.8</version>
         </dependency>
 
+        <!-- excel工具 -->
+        <dependency>
+            <groupId>org.apache.poi</groupId>
+            <artifactId>poi-ooxml</artifactId>
+            <version>4.1.2</version>
+        </dependency>
 
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>mybatis-plus-boot-starter</artifactId>
+            <version>3.5.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>dynamic-datasource-spring-boot-starter</artifactId>
+            <version>2.5.8</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>8.0.33</version>
+        </dependency>
     </dependencies>
 </project>

src/main/java/com/mobai/controller/FluxGetInfoController.java

@@ -1,28 +1,40 @@
 package com.mobai.controller;
 
+import com.mobai.domain.MqttServerModel;
 import com.mobai.domain.Result;
+import com.mobai.domain.VehicleConnectionReq;
 import com.mobai.service.FluxGetInfoService;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
 
 /**
  * @ClassName FluxGetInfo
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/28 22:00
  */
 @RestController
 @Log4j2
-@RequestMapping("fluxmq")
+@RequestMapping("fluxmq/")
 public class FluxGetInfoController {
     @Autowired
     private FluxGetInfoService fluxGetInfoService;
 
-    @GetMapping("getInfo")
+    @GetMapping("/getInfo/")
     public Result getInfo(){
         return fluxGetInfoService.getInfo(null);
     }
+
+    @GetMapping("getIps/")
+    public Result<List<MqttServerModel>> getIps(){
+        return fluxGetInfoService.getIps();
+    }
+
+    @PostMapping("/getIp")
+    public Result vehicleConnection(@RequestBody(required = false) VehicleConnectionReq req){
+        return fluxGetInfoService.vehicleConnection(req);
+    }
 }

src/main/java/com/mobai/controller/StayTimeController.java

@@ -0,0 +1,17 @@
+package com.mobai.controller;
+
+import com.mobai.service.StayTimeService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @ClassName StayTimeController
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/4 9:54
+ */
+@RestController
+public class StayTimeController {
+    @Autowired
+    private StayTimeService stayTimeService;
+}

src/main/java/com/mobai/domain/AcceptToken.java

@@ -5,7 +5,7 @@ import lombok.Data;
 /**
  * @ClassName AcceptToken
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/29 9:18
  */
 @Data

src/main/java/com/mobai/domain/CPUInfo.java

@@ -1,25 +0,0 @@
-package com.mobai.domain;
-
-@lombok.Data
-public class CPUInfo {
-    /**
-     * CPU核数
-     */
-    private long cpuNum;
-    /**
-     * 内核态使用率
-     */
-    private String cSys;
-    /**
-     * 空闲率
-     */
-    private String idle;
-    /**
-     * I/O等待
-     */
-    private String iowait;
-    /**
-     * 用户态使用率
-     */
-    private String user;
-}

src/main/java/com/mobai/domain/Constants.java

@@ -3,7 +3,7 @@ package com.mobai.domain;
 /**
  * @ClassName Constants
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/28 22:05
  */
 public class Constants {

src/main/java/com/mobai/domain/HttpStatus.java

@@ -0,0 +1,93 @@
+package com.mobai.domain;
+
+/**
+ * 返回状态码
+ *
+ * @author ruoyi
+ */
+public class HttpStatus {
+    /**
+     * 操作成功
+     */
+    public static final int SUCCESS = 200;
+
+    /**
+     * 对象创建成功
+     */
+    public static final int CREATED = 201;
+
+    /**
+     * 请求已经被接受
+     */
+    public static final int ACCEPTED = 202;
+
+    /**
+     * 操作已经执行成功，但是没有返回数据
+     */
+    public static final int NO_CONTENT = 204;
+
+    /**
+     * 资源已被移除
+     */
+    public static final int MOVED_PERM = 301;
+
+    /**
+     * 重定向
+     */
+    public static final int SEE_OTHER = 303;
+
+    /**
+     * 资源没有被修改
+     */
+    public static final int NOT_MODIFIED = 304;
+
+    /**
+     * 参数列表错误（缺少，格式不匹配）
+     */
+    public static final int BAD_REQUEST = 400;
+
+    /**
+     * 未授权
+     */
+    public static final int UNAUTHORIZED = 401;
+
+    /**
+     * 访问受限，授权过期
+     */
+    public static final int FORBIDDEN = 403;
+
+    /**
+     * 资源，服务未找到
+     */
+    public static final int NOT_FOUND = 404;
+
+    /**
+     * 不允许的http方法
+     */
+    public static final int BAD_METHOD = 405;
+
+    /**
+     * 资源冲突，或者资源被锁
+     */
+    public static final int CONFLICT = 409;
+
+    /**
+     * 不支持的数据，媒体类型
+     */
+    public static final int UNSUPPORTED_TYPE = 415;
+
+    /**
+     * 系统内部错误
+     */
+    public static final int ERROR = 500;
+
+    /**
+     * 接口未实现
+     */
+    public static final int NOT_IMPLEMENTED = 501;
+
+    /**
+     * 系统警告消息
+     */
+    public static final int WARN = 601;
+}

src/main/java/com/mobai/domain/MqttConnectState.java

@@ -5,7 +5,7 @@ import lombok.Data;
 /**
  * @ClassName MqttConnectState
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/29 10:39
  */
 @Data

src/main/java/com/mobai/domain/MqttInfo.java

@@ -1,52 +0,0 @@
-package com.mobai.domain;
-
-/**
- * MQTT事件信息
- */
-@lombok.Data
-public class MqttInfo {
-    /**
-     * 关闭事件数量
-     */
-    private long closeEventSize;
-    /**
-     * 连接事件数量
-     */
-    private long connectEventSize;
-    /**
-     * 链接总数
-     */
-    private long connectSize;
-    /**
-     * 断开链接数量
-     */
-    private long disconnectEventSize;
-    /**
-     * 推送数量
-     */
-    private long publishEventSize;
-    /**
-     * 发布重试事件数量
-     */
-    private long publishRetryEventSize;
-    /**
-     * 保留消息数量
-     */
-    private long retainSize;
-    /**
-     * 订阅事件数量
-     */
-    private long subscribeEventSize;
-    /**
-     * 订阅数量
-     */
-    private long subscribeSize;
-    /**
-     * 主题数量
-     */
-    private long topicSize;
-    /**
-     * 取消订阅数量
-     */
-    private long unSubscribeEventSize;
-}

src/main/java/com/mobai/domain/MqttServerModel.java

@@ -0,0 +1,28 @@
+package com.mobai.domain;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author DongZl
+ * @description: Mqtt服务器模型
+ * @Date 2024-3-26 上午 09:53
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class MqttServerModel {
+
+    /**
+     * MQTT服务节点
+     */
+    private String broker;
+
+    /**
+     * MQTT订阅主题
+     */
+    private String topic;
+}

src/main/java/com/mobai/domain/Result.java

@@ -1,33 +1,30 @@
 package com.mobai.domain;
 
-import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Data;
-import lombok.NoArgsConstructor;
 
 import java.io.Serializable;
 
 /**
  * 响应信息主体
  *
- * @author mobai
+ * @author ruoyi
  */
 @Data
-@Builder
-@NoArgsConstructor
-@AllArgsConstructor
 public class Result<T> implements Serializable {
     /**
      * 成功
      */
-    public static final int SUCCESS = Constants.SUCCESS;
+    public static final int SUCCESS = HttpStatus.SUCCESS;
     /**
      * 失败
      */
-    public static final int FAIL = Constants.FAIL;
-
-
+    public static final int FAIL = HttpStatus.ERROR;
     private static final long serialVersionUID = 1L;
+    /**
+     * 系统警告消息
+     */
+    private static final int WARN = HttpStatus.WARN;
+
     private int code;
 
     private String msg;
@@ -35,11 +32,11 @@ public class Result<T> implements Serializable {
     private T data;
 
     public static <T> Result<T> success () {
-        return restResult(null, SUCCESS, null);
+        return restResult(null, SUCCESS, "操作成功");
     }
 
     public static <T> Result<T> success (T data) {
-        return restResult(data, SUCCESS, null);
+        return restResult(data, SUCCESS, "操作成功");
     }
 
     public static <T> Result<T> success (T data, String msg) {
@@ -47,7 +44,7 @@ public class Result<T> implements Serializable {
     }
 
     public static <T> Result<T> error () {
-        return restResult(null, FAIL, null);
+        return restResult(null, FAIL, "操作失败");
     }
 
     public static <T> Result<T> error (String msg) {
@@ -55,7 +52,7 @@ public class Result<T> implements Serializable {
     }
 
     public static <T> Result<T> error (T data) {
-        return restResult(data, FAIL, null);
+        return restResult(data, FAIL, "操作失败");
     }
 
     public static <T> Result<T> error (T data, String msg) {
@@ -66,16 +63,32 @@ public class Result<T> implements Serializable {
         return restResult(null, code, msg);
     }
 
+    public static <T> Result<T> warn () {
+        return restResult(null, WARN, "操作失败");
+    }
+
+    public static <T> Result<T> warn (String msg) {
+        return restResult(null, WARN, msg);
+    }
+
+    public static <T> Result<T> warn (T data) {
+        return restResult(data, WARN, "操作失败");
+    }
+
+    public static <T> Result<T> warn (T data, String msg) {
+        return restResult(data, WARN, msg);
+    }
+
     public static <T> Result<T> warn (int code, String msg) {
         return restResult(null, code, msg);
     }
 
     private static <T> Result<T> restResult (T data, int code, String msg) {
-        return Result.<T>builder()
-                .code(code)
-                .data(data)
-                .msg(msg)
-                .build();
+        Result<T> apiResult = new Result<>();
+        apiResult.setCode(code);
+        apiResult.setData(data);
+        apiResult.setMsg(msg);
+        return apiResult;
     }
 
     public static <T> Boolean isError (Result<T> ret) {
@@ -85,5 +98,4 @@ public class Result<T> implements Serializable {
     public static <T> Boolean isSuccess (Result<T> ret) {
         return Result.SUCCESS == ret.getCode();
     }
-
 }

src/main/java/com/mobai/domain/StayTime.java

@@ -0,0 +1,38 @@
+package com.mobai.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Date;
+
+/**
+ * @ClassName StayTime
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/4 9:28
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class StayTime {
+
+    @TableId(value = "id", type = IdType.AUTO)
+    private Long id;
+    private String vin;
+    private String ip;
+    private Long upTime;
+    private Long downTime;
+    private String stayLongTime;
+    private String remark;
+    private String createBy;
+    private Date createTime;
+    private String updateBy;
+    private Date updateTime;
+
+
+}

src/main/java/com/mobai/domain/User.java

@@ -7,7 +7,7 @@ import lombok.NoArgsConstructor;
 /**
  * @ClassName User
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/29 9:05
  */
 @Data

src/main/java/com/mobai/domain/VehicleConnectionReq.java

@@ -0,0 +1,51 @@
+package com.mobai.domain;
+
+import com.alibaba.fastjson2.annotation.JSONField;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author DongZl
+ * @description: 车辆获取连接地址
+ * @Date 2023-11-28 上午 10:32
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class VehicleConnectionReq {
+
+    /**
+     * {
+     *     "vehicleVin": "VIN1234567894",
+     *     "timestamp": "11111",
+     *     "username": "你好",
+     *     "nonce": "33"
+     * }
+     */
+
+    /**
+     * vin
+     */
+    @JSONField(name = "vehicleVin")
+    private String vin;
+
+    /**
+     * 时间戳
+     */
+    private String timestamp;
+
+    /**
+     * 用户名
+     */
+    @JSONField(name = "username")
+    private String userName;
+
+    /**
+     * 随机字符串
+     */
+    private String nonce;
+
+}

src/main/java/com/mobai/domain/flux/ApifoxModel.java

@@ -1,6 +1,6 @@
 // ApifoxModel.java
 
-package com.mobai.domain;
+package com.mobai.domain.flux;
 
 @lombok.Data
 public class ApifoxModel {

src/main/java/com/mobai/domain/flux/CPUInfo.java

@@ -0,0 +1,17 @@
+package com.mobai.domain.flux;
+
+/**
+ * cpu使用率
+ */
+@lombok.Data
+public class CPUInfo {
+    /**
+     * 进程cpu使用率
+     */
+    private String processCpuUsage;
+
+    /**
+     * 系统cpu使用率
+     */
+    private String systemCpuUsage;
+}

src/main/java/com/mobai/domain/flux/Endpoints.java

@@ -0,0 +1,40 @@
+package com.mobai.domain.flux;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * mqtt 结束指针
+ *
+ * @ClassName Endpoints
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/2 15:16
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class Endpoints {
+
+    /**
+     * mqtt 地址
+     */
+    private String mqtt;
+
+    /**
+     * http访问地址
+     */
+    private String http;
+
+    /**
+     * websocket
+     */
+    private String ws;
+
+}
+//            "mqtt":"mqtt://127.0.0.1:1883",
+//            "http":"http://127.0.0.1:8080",
+//            "ws":"ws://127.0.0.1:8883/mqtt"

src/main/java/com/mobai/domain/flux/Events.java

@@ -0,0 +1,70 @@
+package com.mobai.domain.flux;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @ClassName Events
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/2 15:28
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class Events {
+    private Integer mqttConnectEvent;
+    private Integer mqttCloseEvent;
+    private Integer mqttDisconnectEvent;
+    private Integer mqttPublishDropEvent;
+    private Integer mqttPublishUnsubscribeDropEvent;
+    private Integer mqttSubscribeEvent;
+    private Integer mqttUnsubscribeEvent;
+    private Integer mqttPublishQos0Event;
+    private Integer mqttPublishQos1Event;
+    private Integer mqttPublishQos2Event;
+    private Integer mqttWriteQos0Event;
+    private Integer mqttWriteQos1Event;
+    private Integer mqttWriteQos2Event;
+    private Integer mqttRewriteQos1Event;
+    private Integer mqttRewriteQos2Event;
+    private Integer mqttWriteDropEvent;
+    private Integer authSuccessEvent;
+    private Integer authFailedEvent;
+    private Integer bridgeSuccessEvent;
+    private Integer bridgeErrorEvent;
+    private Integer bridgeDropEvent;
+    private Integer eventbusSendSuccessEvent;
+    private Integer eventbusSendErrorEvent;
+    private Integer eventbusReceiveEvent;
+    private Integer eventbusDiscardEvent;
+
+}
+//"mqtt.connect.event": 11,
+//"mqtt.close.event": 9,
+//"mqtt.disconnect.event": 3,
+//"mqtt.publish.drop.event": 27,
+//"mqtt.publish.unsubscribe.drop.event": 27,
+//"mqtt.subscribe.event": 24,
+//"mqtt.unsubscribe.event": 1,
+//"mqtt.publish.qos0.event": 159,
+//"mqtt.publish.qos1.event": 0,
+//"mqtt.publish.qos2.event": 0,
+//"mqtt.write.qos0.event": 133,
+//"mqtt.write.qos1.event": 0,
+//"mqtt.write.qos2.event": 0,
+//"mqtt.rewrite.qos1.event": 0,
+//"mqtt.rewrite.qos2.event": 0,
+//"mqtt.write.drop.event": 0,
+//"auth.success.event": 0,
+//"auth.failed.event": 0,
+//"bridge.success.event": 0,
+//"bridge.error.event": 0,
+//"bridge.drop.event": 0,
+//"eventbus.send.success.event": 0,
+//"eventbus.send.error.event": 0,
+//"eventbus.receive.event": 0,
+//"eventbus.discard.event": 0

src/main/java/com/mobai/domain/flux/FlowInfo.java

@@ -1,4 +1,4 @@
-package com.mobai.domain;
+package com.mobai.domain.flux;
 
 /**
  * 节点状态

src/main/java/com/mobai/domain/flux/JVMInfo.java

@@ -1,4 +1,4 @@
-package com.mobai.domain;
+package com.mobai.domain.flux;
 
 /**
  * JVM使用信息
@@ -53,4 +53,18 @@ public class JVMInfo {
      * 线程数量
      */
     private long threadCount;
+
+    /**
+     * os
+     */
+    private String  osArch;
+    /**
+     * 系统名
+     */
+    private String  osName;
+
+    /**
+     * cpu数量
+     */
+    private String  cpuCount;
 }

src/main/java/com/mobai/domain/flux/MqttInfo.java

@@ -0,0 +1,11 @@
+package com.mobai.domain.flux;
+
+/**
+ * MQTT事件信息
+ */
+@lombok.Data
+public class MqttInfo {
+
+    private RunTimes runtimes;
+    private Events events;
+}

src/main/java/com/mobai/domain/flux/RunTimes.java

@@ -0,0 +1,87 @@
+package com.mobai.domain.flux;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * 运行时数据
+ * @ClassName RunTimes
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/2 15:22
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class RunTimes {
+
+    /**
+     * mqtt 连接数量
+     */
+    private Integer mqttConnect;
+
+    /**
+     * 缓存大小
+     */
+    private Integer retainSize;
+    /**
+     * 主题个数
+     */
+    private Integer topicSize;
+
+    /**
+     * 脚本数量
+     */
+    private Integer aclCount;
+
+    /**
+     * 脚本数量
+     */
+    private Integer scriptCount;
+    /**
+     * 代理订阅数量
+     */
+    private Integer proxySubCount;
+    /**
+     * 消费者数量
+     */
+    private Integer commandConsumerCount;
+    /**
+     * mqtt 订阅数量
+     */
+    private Integer mqttSubscribeQos0;
+    /**
+     * 订阅数量
+     */
+    private Integer mqttSubscribeQos1;
+    /**
+     * 订阅数量
+     */
+    private Integer mqttSubscribeQos2;
+    /**
+     * 规则引擎数量
+     */
+    private Integer ruleCount;
+    /**
+     * 桥接数量
+     */
+    private Integer bridgeOver;
+
+
+//  "mqtt.connect": 2,
+//  "retain.size": 0,
+//  "topic.size": 3,
+//  "acl.count": 0,
+//  "script.count": 0,
+//  "proxySub.count": 0,
+//  "command.consumer.count": 0,
+//  "mqtt.subscribe.qos0": 3,
+//  "mqtt.subscribe.qos1": 0,
+//  "mqtt.subscribe.qos2": 0,
+//  "rule.count": 0,
+//  "bridge.over": 0
+
+}

src/main/java/com/mobai/mapper/StayTimeMapper.java

@@ -0,0 +1,15 @@
+package com.mobai.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.mobai.domain.StayTime;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * @ClassName StayTimeMapper
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/4 9:57
+ */
+@Mapper
+public interface StayTimeMapper extends BaseMapper<StayTime> {
+}

src/main/java/com/mobai/mq/fluxmq/FluxMqInit.java

@@ -7,7 +7,7 @@ import java.io.IOException;
 /**
  * @ClassName FluxMqInit
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/27 9:57
  */
 public class FluxMqInit {

src/main/java/com/mobai/mq/rabbitmq/Custom.java

@@ -1,35 +0,0 @@
-package com.mobai.mq.rabbitmq;
-
-import com.rabbitmq.client.Channel;
-import org.springframework.amqp.core.ExchangeTypes;
-import org.springframework.amqp.core.Message;
-import org.springframework.amqp.rabbit.annotation.Exchange;
-import org.springframework.amqp.rabbit.annotation.Queue;
-import org.springframework.amqp.rabbit.annotation.QueueBinding;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.stereotype.Component;
-
-import java.io.IOException;
-
-/**
- * @ClassName Custom
- * @Description 描述
- * @Author SaiSai.Liu
- * @Date 2024/5/26 15:25
- */
-@Component
-public class Custom {
-
-//    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "send_sms_queue"),
-//            exchange = @Exchange(value = "null",type = ExchangeTypes.DIRECT)))
-//    public void mqCustom(Object data, Message message, Channel channel){
-//        System.out.println(data.toString());
-//        try {
-//            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
-//        } catch (IOException e) {
-//            throw new RuntimeException(e);
-//        }
-//
-//    }
-
-}

src/main/java/com/mobai/mq/rabbitmq/api/MqttPublishSample.java

@@ -0,0 +1,44 @@
+package com.mobai.mq.rabbitmq.api;
+
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+
+public class MqttPublishSample {
+
+    public static void main(String[] args) {
+
+        String topic        = "MQTT Examples";
+        String content      = "Message from MqttPublishSample";
+        int qos             = 2;
+        String broker       = "tcp://iot.eclipse.org:1883";
+        String clientId     = "JavaSample";
+        MemoryPersistence persistence = new MemoryPersistence();
+
+        try {
+            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
+            MqttConnectOptions connOpts = new MqttConnectOptions();
+            connOpts.setCleanSession(true);
+            System.out.println("Connecting to broker: "+broker);
+            sampleClient.connect(connOpts);
+            System.out.println("Connected");
+            System.out.println("Publishing message: "+content);
+            MqttMessage message = new MqttMessage(content.getBytes());
+            message.setQos(qos);
+            sampleClient.publish(topic, message);
+            System.out.println("Message published");
+            sampleClient.disconnect();
+            System.out.println("Disconnected");
+            System.exit(0);
+        } catch(MqttException me) {
+            System.out.println("reason "+me.getReasonCode());
+            System.out.println("msg "+me.getMessage());
+            System.out.println("loc "+me.getLocalizedMessage());
+            System.out.println("cause "+me.getCause());
+            System.out.println("excep "+me);
+            me.printStackTrace();
+        }
+    }
+}

src/main/java/com/mobai/mq/rabbitmq/api/SubscribeSample.java

@@ -0,0 +1,33 @@
+package com.mobai.mq.rabbitmq.api;
+
+import com.mobai.mq.rabbitmq.cofig.MqttFactory;
+import com.mobai.mq.rabbitmq.cofig.MqttProperties;
+import com.mobai.mq.rabbitmq.domian.GetOptions;
+import com.mobai.mq.rabbitmq.domian.MqttCallBackServiceImpl;
+import org.eclipse.paho.client.mqttv3.*;
+
+public class SubscribeSample {
+    public static void main(String[] args) {
+        MqttProperties mqttProperties = MqttProperties.configBuild("39.98.69.92", "topic0");
+//        MqttProperties mqttProperties = new MqttProperties();
+//          mqttProperties.setBroker("tcp://39.98.69.92:1883");
+//          mqttProperties.setTopic("mqtt/test");
+          mqttProperties.setUsername("emqx");
+          mqttProperties.setPassword("public");
+//          mqttProperties.setClientid("subscribe_client");
+
+        int qos = 0;
+        try {
+            MqttClient client = new MqttFactory(new MqttCallBackServiceImpl()).buildOptions(mqttProperties);
+            // 连接参数
+            MqttConnectOptions options = GetOptions.getMqttOptionas(mqttProperties);
+            // 设置回调
+            client.setCallback(new MqttCallBackServiceImpl());
+            // 进行连接
+            client.connect(options);
+            client.subscribe(mqttProperties.getTopic(), qos);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/java/com/mobai/mq/rabbitmq/cofig/MqttFactory.java

@@ -0,0 +1,44 @@
+package com.mobai.mq.rabbitmq.cofig;
+
+import com.mobai.mq.rabbitmq.domian.GetOptions;
+import com.mobai.mq.rabbitmq.domian.MqttCallBackServiceImpl;
+import lombok.AllArgsConstructor;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+/**
+ * @ClassName MqttFactory
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/5/31 11:35
+ */
+
+@Service
+@AllArgsConstructor
+public class MqttFactory {
+
+    private final MqttCallBackServiceImpl mqttCallBackService;
+
+    public MqttClient buildOptions( MqttProperties mqttProperties)  {
+        MqttClient client = null;
+        try {
+            client = new MqttClient(
+                    mqttProperties.getBroker(),
+                    mqttProperties.getClientid(),
+                    new MemoryPersistence());
+            MqttConnectOptions options = GetOptions.getMqttOptionas(mqttProperties);
+            client.connect(options);
+            client.setCallback(mqttCallBackService);
+            // 事件订阅
+            client.subscribe(mqttProperties.getTopic(), 0);
+
+        } catch (MqttException e) {
+            throw new RuntimeException(e);
+        }
+
+        return client;
+    }
+}

src/main/java/com/mobai/mq/rabbitmq/cofig/MqttProperties.java

@@ -0,0 +1,33 @@
+package com.mobai.mq.rabbitmq.cofig;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @ClassName MqttProperties
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/5/30 20:05
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class MqttProperties {
+    private String broker ;
+    private String topic ;
+    private String username;
+    private String password;
+    private String clientid;
+
+    public static MqttProperties configBuild(String ip,String topic){
+//        "tcp://39.98.69.92:1883"
+        return MqttProperties.builder()
+                .broker("tcp://"+ip+":1883")
+                .topic(topic)
+                .clientid("mobai-mq")
+                .build();
+    }
+}

src/main/java/com/mobai/mq/rabbitmq/cofig/RabbitConfig.java

@@ -0,0 +1,46 @@
+package com.mobai.mq.rabbitmq.cofig;
+
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @ClassName RabbitConfig
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/5/31 21:47
+ */
+@Configuration
+public class RabbitConfig {
+
+    @Bean
+    public Queue initQueue(){
+        return new Queue("create.topic",true);
+    }
+
+    /**
+     * 创建主题队列
+     * @return
+     */
+    @Bean
+    public DirectExchange direct() {
+        return new DirectExchange("mobai-mq");
+    }
+
+    /**
+     * 绑定交换机
+     * @param direct
+     * @param initQueue
+     * @return
+     */
+    @Bean
+    public Binding binding1a(DirectExchange direct,
+                             Queue initQueue) {
+        return BindingBuilder.bind(initQueue)
+                .to(direct)
+                .with("mobai-mq");
+    }
+}

src/main/java/com/mobai/mq/rabbitmq/domian/GetOptions.java

@@ -0,0 +1,28 @@
+package com.mobai.mq.rabbitmq.domian;
+
+import com.mobai.mq.rabbitmq.cofig.MqttProperties;
+import org.apache.commons.lang3.StringUtils;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+
+/**
+ * @ClassName GetOptions
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/5/31 14:26
+ */
+
+public class GetOptions {
+
+    public static MqttConnectOptions getMqttOptionas(MqttProperties mqttProperties){
+        // 连接参数
+        MqttConnectOptions options = new MqttConnectOptions();
+        if (!StringUtils.isAllBlank(mqttProperties.getUsername(),
+                mqttProperties.getPassword())){
+            options.setUserName(mqttProperties.getUsername());
+            options.setPassword(mqttProperties.getPassword().toCharArray());
+        }
+        options.setConnectionTimeout(60);
+        options.setKeepAliveInterval(60);
+        return options;
+    }
+}

src/main/java/com/mobai/mq/rabbitmq/domian/MqttCallBackServiceImpl.java

@@ -0,0 +1,35 @@
+package com.mobai.mq.rabbitmq.domian;
+
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.stereotype.Service;
+
+/**
+ * 回执消息类
+ * @ClassName MqttCallBackServiceImpl
+ * @Description 回执消息类
+ * @Author Mobai
+ * @Date 2024/5/30 20:02
+ */
+@Service
+public class MqttCallBackServiceImpl implements MqttCallback {
+    @Override
+    public void connectionLost(Throwable cause) {
+        System.out.println("connectionLost: " + cause.getMessage());
+    }
+
+    @Override
+    public void messageArrived(String topic, MqttMessage message) {
+        System.out.println("topic: " + topic);
+        System.out.println("Qos: " + message.getQos());
+        System.out.println("message content: " + new String(message.getPayload()));
+
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        System.out.println("deliveryComplete---------" + token.isComplete());
+    }
+
+}

src/main/java/com/mobai/mq/rabbitmq/rabbitMq/MessageHandler.java

@@ -0,0 +1,156 @@
+package com.mobai.mq.rabbitmq.rabbitMq;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.mobai.domain.StayTime;
+import com.mobai.mq.rabbitmq.cofig.MqttFactory;
+import com.mobai.service.StayTimeService;
+import com.mobai.util.RedisService;
+import com.rabbitmq.client.Channel;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.servlet.ServletException;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+/**
+ * 消费者：消息处理器
+ *
+ * @ClassName MessageHandler
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/5/31 14:37
+ */
+@Log4j2
+@Component
+public class MessageHandler {
+
+    @Autowired
+    private MqttFactory mqttFactory;
+
+    @Autowired
+    private StayTimeService stayTimeService;
+
+    @Autowired
+    private RedisService redisService;
+
+//    @RabbitListener(queues = {"create.topic"})
+//    private void message(String msg) {
+//        log.info("消息内容：{}", msg);
+////        MqttProperties topic0 = MqttProperties.configBuild(
+////                "39.98.69.92",
+////                "topic0");
+////        log.info("接收到消息：{}", topic0);
+////        MqttClient client = mqttFactory.buildOptions(topic0);
+////        log.info("client创建:{}", client);
+////        log.info("clientID创建:{}", client.getClientId());
+//    }
+
+
+    /**
+     * 上线事件
+     *
+     * @param msg
+     */
+    @RabbitListener(queues = {"event"})
+    private void trainUp(String msg, Message message, Channel channel) {
+        log.info("event:{}", msg);
+        // 事件内容
+        JSONObject jsonObject = JSON.parseObject(msg);
+        String clientId = JSON.to(String.class, jsonObject.get("clientId"));
+        if (!(clientId.contains("-") && clientId.contains("."))) {
+            log.error("不是车辆事件：：{}",message);
+            try {
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
+                return;
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        String[] split = clientId.split("-");
+        String vin = split[0];
+        String ip = split[1];
+        long timestamp = JSON.to(Long.class, jsonObject.get("timestamp"));
+        // 判断登陆事件
+        if (jsonObject.get("auth") != null) {
+            try {
+                log.info("上线事件");
+                log.info("上线车辆vin:{}--上线时ip:{}", vin, ip);
+                if (ip == null) {
+                    throw new ServletException("上线时ip为空");
+                }
+                // 方法内有判断，有则自增，无则创建
+                redisService.increment("onlineCar-" + ip, 1);
+                boolean save = stayTimeService.save(new StayTime() {{
+                    setIp(ip);
+                    setVin(vin);
+                    setUpTime(timestamp);
+                }});
+                log.info(save ? vin + "上线记录成功" : vin + "上线记录失败");
+                // 消息消费成功则确认
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+            } catch (IOException | ServletException e) {
+                log.error("上线失败");
+                try {
+                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
+                } catch (IOException ex) {
+                    throw new RuntimeException(ex);
+                }
+                throw new RuntimeException(e);
+            }
+        } else {
+            log.info("下线事件");
+            try {
+                log.info("下线车辆vin:{}--下线时ip:{}", vin, ip);
+                if (ip == null) {
+                    throw new ServletException("下线时ip为空");
+                }
+                // 方法内有判断，有则自减，无则创建
+                redisService.decrement("onlineCar-" + ip, 1);
+                // 判断是否有该记录
+                StayTime vinStayTime = stayTimeService.getOne(new LambdaQueryWrapper<>() {{
+                    eq(StayTime::getVin, vin);
+                    eq(StayTime::getDownTime, 0);
+                }});
+                // 为空抛异常
+                if (vinStayTime==null){
+                    throw new ServletException("{"+vin+"}数据不存在");
+                }
+                String format = new SimpleDateFormat("HH时mm分ss秒").format(new Date(timestamp - vinStayTime.getUpTime()-8*60*60*1000));
+                vinStayTime.setStayLongTime(format);
+                vinStayTime.setDownTime(timestamp);
+                // 修改下线时间
+                boolean update = stayTimeService.update(vinStayTime, new LambdaUpdateWrapper<>() {{
+                    eq(StayTime::getVin, vin);
+                    eq(StayTime::getDownTime, 0);
+                }});
+                // 输出在线时长
+                log.info("车辆在线时长为：{}", format);
+                log.info(update ? vin + "下线记录成功" : vin + "下线记录失败");
+                redisService.deleteObject(vin);
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+            } catch (ServletException | IOException e) {
+                log.error("下线失败");
+                throw new RuntimeException(e);
+            } finally {
+                try {
+                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
+                } catch (IOException ex) {
+                    throw new RuntimeException(ex);
+                }
+            }
+        }
+
+
+    }
+
+
+}

src/main/java/com/mobai/openApi/SelectInstances.java

@@ -6,9 +6,7 @@ import com.alibaba.fastjson2.JSON;
 import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
 import com.aliyun.ecs20140526.models.DescribeInstancesResponseBody;
 import com.aliyun.tea.TeaException;
-import com.mobai.domain.ApifoxModel;
-import com.mobai.domain.Result;
-import com.mobai.service.FluxGetInfoService;
+import com.mobai.domain.MqttServerModel;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -25,12 +23,11 @@ import java.util.*;
 @Component
 public class SelectInstances {
 
-    @Autowired
-    private FluxGetInfoService fluxGetInfoService;
 
     @Autowired
     private RedisTemplate<String, String> redisTemplate;
 
+
     /**
      * 使用AK&SK初始化账号Client
      *
@@ -50,31 +47,51 @@ public class SelectInstances {
         return new com.aliyun.ecs20140526.Client(config);
     }
 
-    @Scheduled(cron = "0/10 * * * * ? ")
-    public void saveIps() throws Exception {
-        List<String> ips = new ArrayList<>();
-        com.aliyun.ecs20140526.Client client = SelectInstances.createClient();
-        com.aliyun.ecs20140526.models.DescribeInstancesRequest describeInstancesRequest = new com.aliyun.ecs20140526.models.DescribeInstancesRequest()
-//                .setImageId("m-8vb8qnidv34yj3nbirhc")
-                .setRegionId("cn-zhangjiakou");
-        com.aliyun.teautil.models.RuntimeOptions runtime = new com.aliyun.teautil.models.RuntimeOptions();
+
+    // 通过接口获取数据
+    public DescribeInstancesResponse getInfo() {
+        DescribeInstancesResponse describeInstancesResponse = null;
         try {
-            // 复制代码运行请自行打印 API 的返回值
-            DescribeInstancesResponse describeInstancesResponse = client.describeInstancesWithOptions(describeInstancesRequest, runtime);
-            DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstances instances = describeInstancesResponse.getBody().getInstances();
-            List<DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance> instance =
-                    instances.getInstance();
-            for (DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance inst : instance) {
-                List<String> ipAddress = inst
-                        .getPublicIpAddress()
-                        .getIpAddress();
-                if (ipAddress.isEmpty()) {
-                    continue;
-                } else {
-                    ipAddress.forEach(ip -> ips.add(ip));
-                }
+            com.aliyun.ecs20140526.Client client = SelectInstances.createClient();
+            com.aliyun.ecs20140526.models.DescribeInstancesRequest describeInstancesRequest = new com.aliyun.ecs20140526.models.DescribeInstancesRequest()
+//                .setImageId("m-8vb8qnidv34yj3nbirhc")
+                    .setRegionId("cn-zhangjiakou");
+            com.aliyun.teautil.models.RuntimeOptions runtime = new com.aliyun.teautil.models.RuntimeOptions();
+            describeInstancesResponse = client.describeInstancesWithOptions(describeInstancesRequest, runtime);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        return describeInstancesResponse;
+    }
+
+    public List<String> ips(DescribeInstancesResponse response) {
+        List<String> ips = new ArrayList<>();
+        DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstances instances = response.getBody().getInstances();
+        List<DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance> instance =
+                instances.getInstance();
+        for (DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance inst : instance) {
+            List<String> ipAddress = inst
+                    .getPublicIpAddress()
+                    .getIpAddress();
+            if (ipAddress.isEmpty()) {
+                continue;
+            } else {
+                ipAddress.forEach(ip -> ips.add(ip));
             }
-            log.info("当前实例ip为{}", ips);
+        }
+        log.info("当前实例ip为{}", ips);
+        return ips;
+    }
+
+    //1分钟
+    @Scheduled(cron = "* 0/5  * * * ? ")
+    //10秒
+//    @Scheduled(cron = "0/10 * * * * ? ")
+    public void saveIps() {
+        List<String> ips = null;
+        DescribeInstancesResponse response = this.getInfo();
+        try {
+            ips = this.ips(response);
         } catch (TeaException error) {
             // 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
             // 错误 message
@@ -93,14 +110,13 @@ public class SelectInstances {
             } else {
                 log.error("诊断地址：：{}", error.getData().get("Recommend"));
             }
-
             com.aliyun.teautil.Common.assertAsString(error.message);
         }
         List<SmallNode> nodes = new ArrayList<>();
         for (String ip : ips) {
-            Result<ApifoxModel> info = fluxGetInfoService.getInfo(ip);
-            //连接总数
-            long connectSize = info.getData().getMqttInfo().getConnectSize();
+            // 获取连接总数
+            String string = redisTemplate.opsForValue().get("onlineCar-" + ip);
+            long connectSize = Long.parseLong(string == null ? "0" : string);
             log.info("{}::{}", ip, connectSize);
             //添加到一个容器
             nodes.add(new SmallNode(ip, connectSize));
@@ -126,35 +142,39 @@ public class SelectInstances {
             arithmet.put(node.getIp(), value);
         }// 根据权重总和计算每个节点的特定比例
         log.info("总可负载量：{}", arithmet);
-        List<String> ips = new ArrayList<>();
+        List<MqttServerModel> ips = new ArrayList<>();
         //获取每个ip的分配率
         for (String ip : arithmet.keySet()) {
             //概率
             BigDecimal probability = arithmet.get(ip).divide(sum, 4, BigDecimal.ROUND_HALF_UP).multiply(BigDecimal.valueOf(100));
             arithmet.put(ip, probability);
-            log.info("{}可负载率(权重值)：{}",ip, probability);
+            log.info("{}可负载率(权重值)：{}", ip, probability);
         }
         Set<String> ipSet = arithmet.keySet();
-
         BigDecimal finalSum = sum;
         Map<String, Integer> map = new HashMap<>();
         // 转换成数量
-        ipSet.forEach(ip ->
-                map.put(ip, arithmet.get(ip).multiply(finalSum).intValue() / 100)
-        );
+        ipSet.forEach(ip -> map.put(ip, arithmet.get(ip).multiply(finalSum).intValue() / 100));
         Long i = 0L;
-        log.info("ip对应可分配车辆：{}",map);
+        log.info("ip对应可分配车辆：{}", map);
         int sumInit = map.values().stream().mapToInt(num -> num).sum();
+
         while (true) {
             ipSet = map.keySet();
             Iterator<String> iterator = ipSet.iterator();
             i++;
             while (iterator.hasNext()) {
+                MqttServerModel mqttServerModel = new MqttServerModel();
                 String ip = iterator.next();
-                ips.add(ip);
+                for (SmallNode node : nodes) {
+                    if (node.ip.equals(ip)) mqttServerModel = new MqttServerModel(ip, "topic" + nodes.indexOf(node));
+                }
+                ips.add(mqttServerModel);
+
                 int i1 = map.get(ip) - 1;
                 map.put(ip, i1);
-                if (map.get(ip).equals(0)) {
+                if (i1 == 0) {
+                    iterator.remove();
                     map.remove(ip);
                 }
             }
@@ -162,9 +182,10 @@ public class SelectInstances {
                 break;
             }
         }
-        redisTemplate.opsForValue().set("fluxMq", JSON.toJSONString(ips));
         // 可负载IP轮询排列
         log.info("排列ip,{}", ips);
+        Boolean mqttIp = redisTemplate.delete("mqttIp");
+        ips.forEach(mqtt -> redisTemplate.opsForList().leftPush("mqttIp", JSON.toJSONString(mqtt)));
     }
 
 

src/main/java/com/mobai/runner/MqttRunner.java

@@ -0,0 +1,43 @@
+package com.mobai.runner;
+
+import com.alibaba.fastjson.JSON;
+import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
+import com.mobai.domain.MqttServerModel;
+import com.mobai.mq.rabbitmq.cofig.MqttFactory;
+import com.mobai.mq.rabbitmq.cofig.MqttProperties;
+import com.mobai.mq.rabbitmq.domian.GetOptions;
+import com.mobai.mq.rabbitmq.domian.MqttCallBackServiceImpl;
+import com.mobai.openApi.SelectInstances;
+import com.mobai.util.RedisService;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.data.redis.connection.RedisServer;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * @ClassName MqttRunner
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/4 20:03
+ */
+@Component
+public class MqttRunner implements ApplicationRunner {
+
+    @Autowired
+    private SelectInstances selectInstances;
+
+    @Autowired
+    private RedisService redisService;
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        // 存入mqttIp
+        selectInstances.saveIps();
+
+    }
+}

src/main/java/com/mobai/service/FluxGetInfoService.java

@@ -1,15 +1,24 @@
 package com.mobai.service;
 
+import com.mobai.domain.MqttServerModel;
 import com.mobai.domain.Result;
+import com.mobai.domain.VehicleConnectionReq;
+
+import java.util.List;
 
 /**
  * @ClassName FluxGetInfoService
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/28 22:01
  */
 public interface FluxGetInfoService {
 
     Result getInfo(String ip);
 
+
+
+    Result vehicleConnection(VehicleConnectionReq req);
+
+    Result<List<MqttServerModel>> getIps();
 }

src/main/java/com/mobai/service/StayTimeService.java

@@ -0,0 +1,14 @@
+package com.mobai.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.mobai.domain.StayTime;
+import com.mobai.mapper.StayTimeMapper;
+
+/**
+ * @ClassName StayTimeService
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/4 9:54
+ */
+public interface StayTimeService extends IService<StayTime> {
+}

src/main/java/com/mobai/service/impl/FluxGetInfoServiceImpl.java

@@ -1,61 +1,122 @@
 package com.mobai.service.impl;
 
 import com.alibaba.fastjson2.JSON;
-import com.mobai.domain.AcceptToken;
-import com.mobai.domain.ApifoxModel;
-import com.mobai.domain.Result;
-import com.mobai.domain.User;
+import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
+import com.mobai.domain.*;
+import com.mobai.domain.flux.ApifoxModel;
+import com.mobai.openApi.SelectInstances;
 import com.mobai.service.FluxGetInfoService;
-import okhttp3.*;
-import org.apache.catalina.authenticator.SpnegoAuthenticator;
+import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
-import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.client.RestTemplate;
 
-import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * @ClassName FluxGetInfoServiceImpl
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/28 22:01
  */
+
+@Log4j2
 @Service
 public class FluxGetInfoServiceImpl implements FluxGetInfoService {
     @Autowired
     private RestTemplate restTemplate;
 
+    @Autowired
+    private RedisTemplate<String, String> redis;
+
+    @Autowired
+    private SelectInstances selectInstances;
+
+    /**
+     * 通过ip获取详细信息
+     *
+     * @param ip
+     * @return
+     */
     @Override
     public Result getInfo(String ip) {
         String url = null;
-        if (ip==null){
-             url= "http://39.98.50.223:8080/public/";
-        }else {
-            url = "http://"+ip+":8080/public/";
+        if (ip == null) {
+            url = "http://39.98.69.92:8080/public/";
+        } else {
+            url = "http://" + ip + ":8080/public/";
         }
         User user = new User("fluxmq", "fluxmq");
         //登录
-        AcceptToken token = restTemplate.postForObject(url+"login", user, AcceptToken.class);
-
+        AcceptToken token = restTemplate.postForObject(url + "login", user, AcceptToken.class);
         //请求头
         HttpHeaders headers = new HttpHeaders();
         headers.add("token", token.getAccessToken());
         //封装请求头
         HttpEntity<MultiValueMap<String, Object>> formEntity = new HttpEntity<>(headers);
         ResponseEntity<String> exchange = restTemplate.exchange(url + "cluster", HttpMethod.GET, formEntity, String.class);
+        System.out.println(exchange);
 //        System.out.println(exchange);
 //        System.out.println(exchange.getBody());
         List<ApifoxModel> apifoxModel = JSON.parseArray(exchange.getBody(), ApifoxModel.class);
         // get 获取具体所有信息
+        log.info(apifoxModel);
 
         return Result.success(apifoxModel.get(0));
     }
+
+
+    @Override
+    public Result vehicleConnection(VehicleConnectionReq req) {
+        // "vehicleVin": "VIN1234567894"  vin
+        // "timestamp": "11111"           new Date().getMillis()
+        // "username": "你好"              Vin + timestamp
+        // "nonce": "33"                  随机
+        log.warn("参数为：{}", req);
+//        String string = redis.opsForList().range("fluxMq", 0, -1).get(0);
+//        List<MqttServerModel> mqtts = JSON.parseArray(string, MqttServerModel.class);
+//        log.info("集合：{}",mqtts);
+//        tcp://192.168.1.1:1883
+        if (redis.hasKey("fluxMqIndex")) {
+            redis.opsForValue().increment("fluxMqIndex", 1);
+        } else {
+            redis.opsForValue().set("fluxMqIndex", 0 + "");
+        }
+        int index = Integer.valueOf(redis.opsForValue().get("fluxMqIndex"));
+        log.info("下标：{}", index);
+//        List<String> fluxmq = redis.opsForList().range("mqttIp", 0, 5);
+        String mqttIp = redis.opsForList().index("mqttIp", index);
+
+//        log.info(fluxmq);
+        log.info(mqttIp);
+        MqttServerModel mqttServerModel = JSON.parseObject(mqttIp, MqttServerModel.class);
+        if (index + 1 == 100) {
+            redis.opsForValue().set("fluxMqIndex", 0 + "");
+        }
+        log.info("已获取到对象：{}", mqttServerModel);
+        return Result.success(mqttServerModel);
+    }
+
+    @Override
+    public Result<List<MqttServerModel>> getIps() {
+        List<String> ips = new ArrayList<>();
+        DescribeInstancesResponse response = selectInstances.getInfo();
+        ips = selectInstances.ips(response);
+        log.info("当前实例ip为{}", ips);
+        List<String> finalIps = ips;
+        List<MqttServerModel> collect = ips.stream().map(ip ->
+                new MqttServerModel() {{
+                    setBroker(ip);
+                    setTopic("topic" + finalIps.indexOf(ip));
+                }}).collect(Collectors.toList());
+        return Result.success(collect);
+    }
 }

src/main/java/com/mobai/service/impl/StayTimeServiceImpl.java

@@ -0,0 +1,24 @@
+package com.mobai.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.mobai.domain.StayTime;
+import com.mobai.mapper.StayTimeMapper;
+import com.mobai.service.StayTimeService;
+import com.mobai.util.RedisService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * @ClassName StayTimeServiceImpl
+ * @Description 描述
+ * @Author Mobai
+ * @Date 2024/6/4 9:54
+ */
+
+@Service
+public class StayTimeServiceImpl extends ServiceImpl<StayTimeMapper, StayTime> implements StayTimeService{
+    @Autowired
+    private RedisService redisService;
+
+
+}

src/main/java/com/mobai/util/RedisService.java

@@ -0,0 +1,304 @@
+package com.mobai.util;
+
+import com.alibaba.fastjson2.JSON;
+import org.apache.poi.ss.formula.functions.T;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.data.redis.core.*;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * spring redis 工具类
+ *
+ * @author muyu
+ **/
+@SuppressWarnings(value = {"unchecked", "rawtypes"})
+@Component
+@Lazy
+public class RedisService {
+    @Autowired
+    public RedisTemplate redisTemplate;
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+    /**
+     * 缓存基本的对象，Integer、String、实体类等
+     *
+     * @param key   缓存的键值
+     * @param value 缓存的值
+     */
+    public <T> void setCacheObject(final String key, final T value) {
+        redisTemplate.opsForValue().set(key, value);
+    }
+
+    /**
+     * 缓存基本的对象，Integer、String、实体类等
+     *
+     * @param key      缓存的键值
+     * @param value    缓存的值
+     * @param timeout  时间
+     * @param timeUnit 时间颗粒度
+     */
+    public <T> void setCacheObject(final String key, final T value, final Long timeout, final TimeUnit timeUnit) {
+        redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
+    }
+
+    /**
+     * 设置有效时间
+     *
+     * @param key     Redis键
+     * @param timeout 超时时间
+     * @return true=设置成功；false=设置失败
+     */
+    public boolean expire(final String key, final long timeout) {
+        return expire(key, timeout, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 设置有效时间
+     *
+     * @param key     Redis键
+     * @param timeout 超时时间
+     * @param unit    时间单位
+     * @return true=设置成功；false=设置失败
+     */
+    public boolean expire(final String key, final long timeout, final TimeUnit unit) {
+        return redisTemplate.expire(key, timeout, unit);
+    }
+
+    /**
+     * 获取有效时间
+     *
+     * @param key Redis键
+     * @return 有效时间
+     */
+    public long getExpire(final String key) {
+        return redisTemplate.getExpire(key);
+    }
+
+    /**
+     * 判断 key是否存在
+     *
+     * @param key 键
+     * @return true 存在 false不存在
+     */
+    public Boolean hasKey(String key) {
+        return redisTemplate.hasKey(key);
+    }
+
+    /**
+     * 获得缓存的基本对象。
+     *
+     * @param key 缓存键值
+     * @return 缓存键值对应的数据
+     */
+    public <T> T getCacheObject(final String key) {
+        ValueOperations<String, T> operation = redisTemplate.opsForValue();
+        return operation.get(key);
+    }
+
+    /**
+     * 删除单个对象
+     *
+     * @param key
+     */
+    public boolean deleteObject(final String key) {
+        return redisTemplate.delete(key);
+    }
+
+    /**
+     * 删除集合对象
+     *
+     * @param collection 多个对象
+     * @return
+     */
+    public boolean deleteObject(final Collection collection) {
+        return redisTemplate.delete(collection) > 0;
+    }
+
+    /**
+     * 缓存List数据
+     *
+     * @param key      缓存的键值
+     * @param dataList 待缓存的List数据
+     * @return 缓存的对象
+     */
+    public <T> long setCacheList(final String key, final List<T> dataList) {
+        Long count = redisTemplate.opsForList().rightPushAll(key, dataList);
+        return count == null ? 0 : count;
+    }
+
+    /**
+     * 获得缓存的list对象
+     *
+     * @param key 缓存的键值
+     * @return 缓存键值对应的数据
+     */
+    public <T> List<T> getCacheList(final String key) {
+        return redisTemplate.opsForList().range(key, 0, -1);
+    }
+
+    /**
+     * 缓存Set
+     *
+     * @param key     缓存键值
+     * @param dataSet 缓存的数据
+     * @return 缓存数据的对象
+     */
+    public <T> BoundSetOperations<String, T> setCacheSet(final String key, final Set<T> dataSet) {
+        BoundSetOperations<String, T> setOperation = redisTemplate.boundSetOps(key);
+        Iterator<T> it = dataSet.iterator();
+        while (it.hasNext()) {
+            setOperation.add(it.next());
+        }
+        return setOperation;
+    }
+
+    /**
+     * 获得缓存的set
+     *
+     * @param key
+     * @return
+     */
+    public <T> Set<T> getCacheSet(final String key) {
+        return redisTemplate.opsForSet().members(key);
+    }
+
+    /**
+     * 缓存Map
+     *
+     * @param key
+     * @param dataMap
+     */
+    public <T> void setCacheMap(final String key, final Map<String, T> dataMap) {
+        if (dataMap != null) {
+            redisTemplate.opsForHash().putAll(key, dataMap);
+        }
+    }
+
+    /**
+     * 获得缓存的Map
+     *
+     * @param key
+     * @return
+     */
+    public <T> Map<String, T> getCacheMap(final String key) {
+        return redisTemplate.opsForHash().entries(key);
+    }
+
+    /**
+     * 往Hash中存入数据
+     *
+     * @param key   Redis键
+     * @param hKey  Hash键
+     * @param value 值
+     */
+    public <T> void setCacheMapValue(final String key, final String hKey, final T value) {
+        redisTemplate.opsForHash().put(key, hKey, value);
+    }
+
+    /**
+     * 获取Hash中的数据
+     *
+     * @param key  Redis键
+     * @param hKey Hash键
+     * @return Hash中的对象
+     */
+    public <T> T getCacheMapValue(final String key, final String hKey) {
+        HashOperations<String, String, T> opsForHash = redisTemplate.opsForHash();
+        return opsForHash.get(key, hKey);
+    }
+
+    /**
+     * 获取多个Hash中的数据
+     *
+     * @param key   Redis键
+     * @param hKeys Hash键集合
+     * @return Hash对象集合
+     */
+    public <T> List<T> getMultiCacheMapValue(final String key, final Collection<Object> hKeys) {
+        return redisTemplate.opsForHash().multiGet(key, hKeys);
+    }
+
+    /**
+     * 删除Hash中的某条数据
+     *
+     * @param key  Redis键
+     * @param hKey Hash键
+     * @return 是否成功
+     */
+    public boolean deleteCacheMapValue(final String key, final String hKey) {
+        return redisTemplate.opsForHash().delete(key, hKey) > 0;
+    }
+
+    /**
+     * 获得缓存的基本对象列表
+     *
+     * @param pattern 字符串前缀
+     * @return 对象列表
+     */
+    public Collection<String> keys(final String pattern) {
+        return redisTemplate.keys(pattern);
+    }
+
+    /**
+     * 存入一个集合
+     * @param key
+     * @param t
+     * @param <V>
+     */
+    public <V extends List<?>> void setCacheList(String key, T t) {
+        redisTemplate.opsForList().leftPush(String.valueOf(key),t);
+    }
+
+    public <T> long setCacheList(final String key, final T dataList) {
+        Long count = redisTemplate.opsForList().rightPushAll(key, dataList);
+        return count == null ? 0 : count;
+    }
+
+    public void increment(String s) {
+        if (Boolean.TRUE.equals(stringRedisTemplate.hasKey(s))){
+            stringRedisTemplate.opsForValue().increment(s,1);
+        }else {
+            stringRedisTemplate.opsForValue().set(s,1+"");
+        }
+    }
+
+    public void increment(String s,int i) {
+        if (Boolean.TRUE.equals(stringRedisTemplate.hasKey(s))){
+            String o =  stringRedisTemplate.opsForValue().get(s);
+            System.out.println(o);
+            stringRedisTemplate.opsForValue().increment(s,i);
+        }else {
+            stringRedisTemplate.opsForValue().set(s,i+"");
+        }
+    }
+    public void decrement(String s) {
+        if (Boolean.TRUE.equals(stringRedisTemplate.hasKey(s))){
+            stringRedisTemplate.opsForValue().decrement(s,1);
+        }else {
+            stringRedisTemplate.opsForValue().set(s,0+"");
+        }
+    }
+
+    public void decrement(String s,int i) {
+        if (Boolean.TRUE.equals(stringRedisTemplate.hasKey(s))){
+            stringRedisTemplate.opsForValue().decrement(s,i);
+            String o = (String) stringRedisTemplate.opsForValue().get(s);
+            System.out.println(o);
+        }else {
+            stringRedisTemplate.opsForValue().set(s,0+"");
+        }
+    }
+
+    public String getValue(String vin) {
+        return  stringRedisTemplate.opsForValue().get(vin);
+    }
+
+    public List<String> getList(String key) {
+        return redisTemplate.opsForList().range(key,0,-1);
+    }
+}

src/main/resources/application.yml

@@ -1,5 +1,36 @@
+server:
+  port: 8081
 spring:
   redis:
-    host: 43.142.100.73
+    host: 175.24.138.82
     port: 6379
     password:
+  rabbitmq:
+    host: 175.24.138.82
+    stream:
+      username: guest
+      password: guest
+  datasource:
+    dynamic:
+      primary: mybatis #设置默认的数据源或者数据源组,默认值即为master
+      strict: false #严格匹配数据源,默认false. true未匹配到指定数据源时抛异常,false使用默认数据源
+      datasource:
+#        gmall:
+#          url: jdbc:mysql://hadoop104:3306/gmall?useSSL=false&useUnicode=true&characterEncoding=UTF-8
+#          username: root
+#          password: "000000"
+#          driver-class-name: com.mysql.cj.jdbc.Driver
+        mybatis:
+          url: jdbc:mysql://127.0.0.1:3306/train_working?useSSL=false&useUnicode=true&characterEncoding=UTF-8
+          username: root
+          password: 1234
+          driver-class-name: com.mysql.cj.jdbc.Driver
+# mybatis配置
+mybatis:
+  # 搜索指定包别名
+  typeAliasesPackage: com.mobai
+  # 配置mapper的扫描，找到所有的mapper.xml映射文件
+  mapperLocations: classpath:mapper/**/*.xml
+logging:
+  level:
+    com.mobai.mapper: DEBUG

src/main/resources/mapper/StayTimeMapper.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper
+        PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
+        "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="com.mobai.mapper.StayTimeMapper">
+
+</mapper>

src/test/java/LoginTest.java

@@ -5,7 +5,7 @@ import java.io.IOException;
 /**
  * @ClassName LoginTest
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/28 15:37
  */
 public class LoginTest {

src/test/java/TestGet.java

@@ -5,7 +5,7 @@ import java.io.IOException;
 /**
  * @ClassName TestGet
  * @Description 描述
- * @Author SaiSai.Liu
+ * @Author Mobai
  * @Date 2024/5/28 15:34
  */
 public class TestGet {