feat():before testRunner
parent
a99369e58b
commit
a412e584db
|
|
@ -6,12 +6,10 @@ 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.mq.rabbitmq.cofig.MqttProperties;
|
||||
import com.mobai.service.StayTimeService;
|
||||
import com.mobai.util.RedisService;
|
||||
import com.rabbitmq.client.Channel;
|
||||
import lombok.extern.log4j.Log4j2;
|
||||
import org.eclipse.paho.client.mqttv3.MqttClient;
|
||||
import org.springframework.amqp.core.Message;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
|
|
@ -67,7 +65,7 @@ public class MessageHandler {
|
|||
// 事件内容
|
||||
JSONObject jsonObject = JSON.parseObject(msg);
|
||||
String clientId = JSON.to(String.class, jsonObject.get("clientId"));
|
||||
if (!clientId.contains("VIN")) {
|
||||
if (!(clientId.contains("-") && clientId.contains("."))) {
|
||||
log.error("不是车辆事件：：{}",message);
|
||||
try {
|
||||
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
|
||||
|
|
@ -76,18 +74,7 @@ public class MessageHandler {
|
|||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
if (!clientId.contains("-")){
|
||||
try {
|
||||
throw new ServletException("不是vin——ip");
|
||||
} catch (ServletException e) {
|
||||
try {
|
||||
channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
|
||||
} catch (IOException ex) {
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
String[] split = clientId.split("-");
|
||||
String vin = split[0];
|
||||
String ip = split[1];
|
||||
|
|
|
|||
|
|
@ -7,9 +7,6 @@ import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
|
|||
import com.aliyun.ecs20140526.models.DescribeInstancesResponseBody;
|
||||
import com.aliyun.tea.TeaException;
|
||||
import com.mobai.domain.MqttServerModel;
|
||||
import com.mobai.domain.Result;
|
||||
import com.mobai.domain.flux.ApifoxModel;
|
||||
import com.mobai.service.FluxGetInfoService;
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
|
@ -67,7 +64,7 @@ public class SelectInstances {
|
|||
return describeInstancesResponse;
|
||||
}
|
||||
|
||||
public List<String> ips( DescribeInstancesResponse response){
|
||||
public List<String> ips(DescribeInstancesResponse response) {
|
||||
List<String> ips = new ArrayList<>();
|
||||
DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstances instances = response.getBody().getInstances();
|
||||
List<DescribeInstancesResponseBody.DescribeInstancesResponseBodyInstancesInstance> instance =
|
||||
|
|
@ -113,7 +110,6 @@ public class SelectInstances {
|
|||
} else {
|
||||
log.error("诊断地址：：{}", error.getData().get("Recommend"));
|
||||
}
|
||||
|
||||
com.aliyun.teautil.Common.assertAsString(error.message);
|
||||
}
|
||||
List<SmallNode> nodes = new ArrayList<>();
|
||||
|
|
@ -155,7 +151,6 @@ public class SelectInstances {
|
|||
log.info("{}可负载率(权重值)：{}", ip, probability);
|
||||
}
|
||||
Set<String> ipSet = arithmet.keySet();
|
||||
|
||||
BigDecimal finalSum = sum;
|
||||
Map<String, Integer> map = new HashMap<>();
|
||||
// 转换成数量
|
||||
|
|
@ -191,7 +186,6 @@ public class SelectInstances {
|
|||
log.info("排列ip,{}", ips);
|
||||
Boolean mqttIp = redisTemplate.delete("mqttIp");
|
||||
ips.forEach(mqtt -> redisTemplate.opsForList().leftPush("mqttIp", JSON.toJSONString(mqtt)));
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -39,30 +39,5 @@ public class MqttRunner implements ApplicationRunner {
|
|||
// 存入mqttIp
|
||||
selectInstances.saveIps();
|
||||
|
||||
// List<String> ips = redisService.getList("mqttIp");
|
||||
// List<MqttServerModel> list = ips.stream().distinct().map(str -> JSON.parseObject(str, MqttServerModel.class)).toList();
|
||||
// list.forEach(mqttServerModel -> {
|
||||
// MqttProperties mqttProperties = MqttProperties.configBuild(mqttServerModel.getBroker(), mqttServerModel.getTopic());
|
||||
//// 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,7 +2,6 @@ package com.mobai.service.impl;
|
|||
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
|
||||
import com.aliyun.ecs20140526.models.DescribeInstancesResponseBody;
|
||||
import com.mobai.domain.*;
|
||||
import com.mobai.domain.flux.ApifoxModel;
|
||||
import com.mobai.openApi.SelectInstances;
|
||||
|
|
@ -40,6 +39,7 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
|
|||
|
||||
@Autowired
|
||||
private SelectInstances selectInstances;
|
||||
|
||||
/**
|
||||
* 通过ip获取详细信息
|
||||
*
|
||||
|
|
@ -84,23 +84,24 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
|
|||
// 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
|
||||
// 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);
|
||||
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 == redis.opsForList().size("fluxMq")) {
|
||||
redis.delete("fluxMqIndex");
|
||||
MqttServerModel mqttServerModel = JSON.parseObject(mqttIp, MqttServerModel.class);
|
||||
if (index + 1 == 100) {
|
||||
redis.opsForValue().set("fluxMqIndex", 0 + "");
|
||||
}
|
||||
log.info("已获取到对象：{}",mqttServerModel);
|
||||
log.info("已获取到对象：{}", mqttServerModel);
|
||||
return Result.success(mqttServerModel);
|
||||
}
|
||||
|
||||
|
|
@ -119,5 +120,3 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
|
|||
return Result.success(collect);
|
||||
}
|
||||
}
|
||||
|
||||
// 达到60%开启新服务，30%关闭低实例
|
||||
|
|
|
|||
|
|
@ -2,11 +2,11 @@ server:
|
|||
port: 8081
|
||||
spring:
|
||||
redis:
|
||||
host: 43.142.100.73
|
||||
host: 175.24.138.82
|
||||
port: 6379
|
||||
password:
|
||||
rabbitmq:
|
||||
host: 43.142.100.73
|
||||
host: 175.24.138.82
|
||||
stream:
|
||||
username: guest
|
||||
password: guest
|
||||
|
|
|
|||
Loading…
Reference in New Issue