Mobai
/
TrainNet
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat():更新reids存储节点ip

master
Saisai Liu 2024-06-04 21:25:47 +08:00
parent bd973ab220
commit c997b75228
26 changed files with 101 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/mobai/controller/FluxGetInfoController.java
View File

@ -11,7 +11,7 @@ import org.springframework.web.bind.annotation.*;
/**
* @ClassName FluxGetInfo
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/28 22:00
*/
@RestController

2
src/main/java/com/mobai/controller/StayTimeController.java
View File

@ -7,7 +7,7 @@ import org.springframework.web.bind.annotation.RestController;
/**
* @ClassName StayTimeController
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/4 9:54
*/
@RestController

2
src/main/java/com/mobai/domain/AcceptToken.java
View File

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

2
src/main/java/com/mobai/domain/Constants.java
View File

@ -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 {

2
src/main/java/com/mobai/domain/MqttConnectState.java
View File

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

2
src/main/java/com/mobai/domain/StayTime.java
View File

@ -12,7 +12,7 @@ import java.util.Date;
/**
* @ClassName StayTime
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/4 9:28
*/
@Data

2
src/main/java/com/mobai/domain/User.java
View File

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

2
src/main/java/com/mobai/domain/flux/Endpoints.java
View File

@ -10,7 +10,7 @@ import lombok.NoArgsConstructor;
*
* @ClassName Endpoints
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/2 15:16
*/
@Data

2
src/main/java/com/mobai/domain/flux/Events.java
View File

@ -8,7 +8,7 @@ import lombok.NoArgsConstructor;
/**
* @ClassName Events
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/2 15:28
*/
@Data

2
src/main/java/com/mobai/domain/flux/RunTimes.java
View File

@ -9,7 +9,7 @@ import lombok.NoArgsConstructor;
* 运行时数据
* @ClassName RunTimes
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/2 15:22
*/
@Data

2
src/main/java/com/mobai/mapper/StayTimeMapper.java
View File

@ -7,7 +7,7 @@ import org.apache.ibatis.annotations.Mapper;
/**
* @ClassName StayTimeMapper
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/4 9:57
*/
@Mapper

2
src/main/java/com/mobai/mq/fluxmq/FluxMqInit.java
View File

@ -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 {

2
src/main/java/com/mobai/mq/rabbitmq/cofig/MqttFactory.java
View File

@ -12,7 +12,7 @@ import org.springframework.stereotype.Service;
/**
* @ClassName MqttFactory
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/31 11:35
*/

2
src/main/java/com/mobai/mq/rabbitmq/cofig/MqttProperties.java
View File

@ -8,7 +8,7 @@ import lombok.NoArgsConstructor;
/**
* @ClassName MqttProperties
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/30 20:05
*/
@Data

2
src/main/java/com/mobai/mq/rabbitmq/cofig/RabbitConfig.java
View File

@ -10,7 +10,7 @@ import org.springframework.context.annotation.Configuration;
/**
* @ClassName RabbitConfig
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/31 21:47
*/
@Configuration

2
src/main/java/com/mobai/mq/rabbitmq/domian/GetOptions.java
View File

@ -7,7 +7,7 @@ import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
/**
* @ClassName GetOptions
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/31 14:26
*/

2
src/main/java/com/mobai/mq/rabbitmq/domian/MqttCallBackServiceImpl.java
View File

@ -9,7 +9,7 @@ import org.springframework.stereotype.Service;
* 回执消息类
* @ClassName MqttCallBackServiceImpl
* @Description 回执消息类
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/30 20:02
*/
@Service

36
src/main/java/com/mobai/mq/rabbitmq/rabbitMq/MessageHandler.java
View File

@ -27,7 +27,7 @@ import java.util.Date;
*
* @ClassName MessageHandler
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/31 14:37
*/
@Log4j2
@ -64,22 +64,7 @@ public class MessageHandler {
@RabbitListener(queues = {"event"})
private void trainUp(String msg, Message message, Channel channel) {
log.info("event:{}", msg);
// 链接事件
// event:{
// "protocol":"MQTT",
// "clientIp":"39.144.107.165",
// "nodeIp":"127.0.0.1",
// "clientId":"VIN123456789DIJE4",
// "version":"MQTT_3_1_1",
// "keepalive":20,
// "cleanSession":true,
// "timestamp":1717466764797,
// "auth":{
// "username":"6D7A546314155D43A339EE4C0410613D86C821299316ADECDB871E08",
// "password":"VklOMTIzNDU2Nzg5RElKRTQxNzE3NDY2NzY1MDg3NTgyNDI4QThEQjA0RkU2OTkzNTM5NDIyNTQ2ODIwQzFFNzc3NDUzQTA4NzIzRTU4NUQyNDRBNjY="
// },
// "messageId":0}
// 事件内容
JSONObject jsonObject = JSON.parseObject(msg);
String vin = JSON.to(String.class, jsonObject.get("clientId"));
long timestamp = JSON.to(Long.class, jsonObject.get("timestamp"));
@ -111,16 +96,6 @@ public class MessageHandler {
throw new RuntimeException(e);
}
} else {
// event:{
// "protocol":"MQTT",
// "messageId":0,
// "timestamp":1717466777367,
// "reason":"normal",
// "clientId":"VIN123456789DIJE4",
// "nodeIp":"127.0.0.1",
// "id":354,
// "clientIp":"39.144.107.165"
// }
log.info("下线事件");
String ip = redisService.getValue(vin);
try {
@ -135,7 +110,11 @@ public class MessageHandler {
eq(StayTime::getVin, vin);
eq(StayTime::getDownTime, 0);
}});
String format = new SimpleDateFormat("HH时mm分ss秒").format(new Date(vinStayTime.getUpTime() - timestamp));
// 为空抛异常
if (vinStayTime==null){
throw new ServletException("{"+vin+"}数据不存在");
}
String format = new SimpleDateFormat("HH时mm分ss秒").format(new Date(timestamp - vinStayTime.getUpTime()));
vinStayTime.setStayLongTime(format);
vinStayTime.setDownTime(timestamp);
// 修改下线时间
@ -150,7 +129,6 @@ public class MessageHandler {
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (ServletException | IOException e) {
log.error("下线失败");
throw new RuntimeException(e);
} finally {
try {

18
src/main/java/com/mobai/openApi/SelectInstances.java
View File

@ -21,7 +21,6 @@ import org.springframework.stereotype.Component;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;
@Log4j2
@Component
@ -33,6 +32,7 @@ public class SelectInstances {
@Autowired
private RedisTemplate<String, String> redisTemplate;
/**
* 使用AK&SK初始化账号Client
*
@ -70,22 +70,14 @@ public class SelectInstances {
return describeInstancesResponse;
}
//10分钟
// @Scheduled(cron = "0 0/5 * * * ? ")
@Scheduled(cron = "0 0/1 * * * ? ")
//10秒
@Scheduled(cron = "0/10 * * * * ? ")
// @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();
DescribeInstancesResponse response = this.getInfo();
try {
// 复制代码运行请自行打印 API 的返回值
// DescribeInstancesResponse describeInstancesResponse = client.describeInstancesWithOptions(describeInstancesRequest, runtime);
DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstances instances = response.getBody().getInstances();
List<DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance> instance =
instances.getInstance();
@ -181,6 +173,7 @@ public class SelectInstances {
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 (i1 == 0) {
@ -195,6 +188,9 @@ public class SelectInstances {
redisTemplate.opsForList().leftPush("fluxMq", JSON.toJSONString(ips));
// 可负载IP轮询排列
log.info("排列ip,{}", ips);
Boolean mqttIp = redisTemplate.delete("mqttIp");
ips.forEach(mqtt->redisTemplate.opsForList().leftPush("mqttIp", JSON.toJSONString(mqtt)));
}

55
src/main/java/com/mobai/runner/MqttRunner.java 100644
View File

@ -0,0 +1,55 @@
//package com.mobai.runner;
//
//import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
//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 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;
//
///**
// * @ClassName MqttRunner
// * @Description 描述
// * @Author Mobai
// * @Date 2024/6/4 20:03
// */
//@Component
//public class MqttRunner implements ApplicationRunner {
//
// @Autowired
// private SelectInstances selectInstances;
// @Override
// public void run(ApplicationArguments args) throws Exception {
// String string = redisServer.get("fluxMq");
//
// 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();
// }
// }
//}

2
src/main/java/com/mobai/service/FluxGetInfoService.java
View File

@ -6,7 +6,7 @@ import com.mobai.domain.VehicleConnectionReq;
/**
* @ClassName FluxGetInfoService
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/28 22:01
*/
public interface FluxGetInfoService {

2
src/main/java/com/mobai/service/StayTimeService.java
View File

@ -7,7 +7,7 @@ import com.mobai.mapper.StayTimeMapper;
/**
* @ClassName StayTimeService
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/4 9:54
*/
public interface StayTimeService extends IService<StayTime> {

17
src/main/java/com/mobai/service/impl/FluxGetInfoServiceImpl.java
View File

@ -20,7 +20,7 @@ import java.util.List;
/**
* @ClassName FluxGetInfoServiceImpl
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/5/28 22:01
*/
@ -74,9 +74,9 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
// "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);
// 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);
@ -85,14 +85,17 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
}
int index = Integer.valueOf(redis.opsForValue().get("fluxMqIndex"));
log.info("下标：{}",index);
MqttServerModel mqttServerModel = mqtts.get(index);
if (index + 1 == redis.opsForList().size("fluxmq")) {
// 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 == redis.opsForList().size("fluxMq")) {
redis.delete("fluxMqIndex");
}
log.info("已获取到对象：{}",mqttServerModel);
return Result.success(mqttServerModel);
}
}
// 达到60%开启新服务，30%关闭低实例

2
src/main/java/com/mobai/service/impl/StayTimeServiceImpl.java
View File

@ -11,7 +11,7 @@ import org.springframework.stereotype.Service;
/**
* @ClassName StayTimeServiceImpl
* @Description 描述
* @Author SaiSai.Liu
* @Author Mobai
* @Date 2024/6/4 9:54
*/

2
src/test/java/LoginTest.java
View File

@ -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 {

2
src/test/java/TestGet.java
View File

@ -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 {