HuangBuZhangTeam
/
mq-load
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat():获取服务器信息 

version-2():添加游标，通过游标+1轮询IP列表，
下一步存入节点信息
master
Saisai Liu 2024-05-31 22:15:44 +08:00
parent 0a9d3c25a8
commit abfc6d8a91
10 changed files with 284 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pom.xml
View File

@ -55,6 +55,7 @@
<version>2.0.46</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>

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

@ -1,12 +1,12 @@
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.*;
/**
* @ClassName FluxGetInfo
@ -16,7 +16,7 @@ import org.springframework.web.bind.annotation.RestController;
*/
@RestController
@Log4j2
@RequestMapping("fluxmq")
@RequestMapping("fluxmq/")
public class FluxGetInfoController {
@Autowired
private FluxGetInfoService fluxGetInfoService;
@ -25,4 +25,9 @@ public class FluxGetInfoController {
public Result getInfo(){
return fluxGetInfoService.getInfo(null);
}
@PostMapping("/getIp")
public Result vehicleConnection(@RequestBody(required = false) VehicleConnectionReq req){
return fluxGetInfoService.vehicleConnection(req);
}
}

93
src/main/java/com/mobai/domain/HttpStatus.java 100644
View File

@ -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;
}

28
src/main/java/com/mobai/domain/MqttServerModel.java 100644
View File

@ -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;
}

54
src/main/java/com/mobai/domain/Result.java
View File

@ -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();
}
}

51
src/main/java/com/mobai/domain/VehicleConnectionReq.java 100644
View File

@ -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;
}

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

@ -7,6 +7,7 @@ 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.MqttServerModel;
import com.mobai.domain.Result;
import com.mobai.service.FluxGetInfoService;
import lombok.AllArgsConstructor;
@ -50,6 +51,7 @@ public class SelectInstances {
return new com.aliyun.ecs20140526.Client(config);
}
// @Scheduled(cron = "0 0/10 * * * ? ")
@Scheduled(cron = "0/10 * * * * ? ")
public void saveIps() throws Exception {
List<String> ips = new ArrayList<>();
@ -126,7 +128,7 @@ 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()) {
//概率
@ -139,22 +141,25 @@ public class SelectInstances {
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);
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) {
map.remove(ip);
}
}
@ -162,7 +167,7 @@ public class SelectInstances {
break;
}
}
redisTemplate.opsForValue().set("fluxMq", JSON.toJSONString(ips));
redisTemplate.opsForList().leftPush("fluxMq", JSON.toJSONString(ips));
// 可负载IP轮询排列
log.info("排列ip,{}", ips);
}

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

@ -1,6 +1,7 @@
package com.mobai.service;
import com.mobai.domain.Result;
import com.mobai.domain.VehicleConnectionReq;
/**
* @ClassName FluxGetInfoService
@ -12,4 +13,8 @@ public interface FluxGetInfoService {
Result getInfo(String ip);
Result vehicleConnection(VehicleConnectionReq req);
}

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

@ -1,26 +1,20 @@
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.mobai.domain.*;
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.List;
import java.util.Map;
/**
* @ClassName FluxGetInfoServiceImpl
@ -28,11 +22,22 @@ import java.util.Map;
* @Author SaiSai.Liu
* @Date 2024/5/28 22:01
*/
@Log4j2
@Service
public class FluxGetInfoServiceImpl implements FluxGetInfoService {
@Autowired
private RestTemplate restTemplate;
@Autowired
private RedisTemplate<String, String> redis;
/**
* 通过ip获取详细信息
*
* @param ip
* @return
*/
@Override
public Result getInfo(String ip) {
String url = null;
@ -44,7 +49,6 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
User user = new User("fluxmq", "fluxmq");
//登录
AcceptToken token = restTemplate.postForObject(url + "login", user, AcceptToken.class);
//请求头
HttpHeaders headers = new HttpHeaders();
headers.add("token", token.getAccessToken());
@ -58,4 +62,33 @@ public class FluxGetInfoServiceImpl implements FluxGetInfoService {
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);
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);
MqttServerModel mqttServerModel = mqtts.get(index);
if (index + 1 == redis.opsForList().size("fluxmq")) {
redis.delete("fluxMqIndex");
}
log.info("已获取到对象：{}",mqttServerModel);
return Result.success(mqttServerModel);
}
}
// 达到60%开启新服务，30%关闭低实例

2
src/main/resources/application.yml
View File

@ -1,3 +1,5 @@
server:
port: 8081
spring:
redis:
host: 43.142.100.73