Three_Group
/
Load_Center
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat:负载中心

master
Yunfei Du 2024-05-31 11:03:46 +08:00
commit 77ade6288e
32 changed files with 1547 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
.gitignore vendored 100644
View File

@ -0,0 +1,38 @@
target/
!.mvn/wrapper/maven-wrapper.jar
!**/src/main/**/target/
!**/src/test/**/target/
### IntelliJ IDEA ###
.idea/modules.xml
.idea/jarRepositories.xml
.idea/compiler.xml
.idea/libraries/
*.iws
*.iml
*.ipr
### Eclipse ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
.sts4-cache
### NetBeans ###
/nbproject/private/
/nbbuild/
/dist/
/nbdist/
/.nb-gradle/
build/
!**/src/main/**/build/
!**/src/test/**/build/
### VS Code ###
.vscode/
### Mac OS ###
.DS_Store

8
.idea/.gitignore vendored 100644
View File

@ -0,0 +1,8 @@
# 默认忽略的文件
/shelf/
/workspace.xml
# 基于编辑器的 HTTP 客户端请求
/httpRequests/
# Datasource local storage ignored files
/dataSources/
/dataSources.local.xml

8
.idea/encodings.xml 100644
View File

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Encoding" defaultCharsetForPropertiesFiles="UTF-8">
<file url="file://$PROJECT_DIR$/src/main/java" charset="UTF-8" />
<file url="file://$PROJECT_DIR$/src/main/resources" charset="UTF-8" />
<file url="PROJECT" charset="UTF-8" />
</component>
</project>

5
.idea/inspectionProfiles/Project_Default.xml 100644
View File

@ -0,0 +1,5 @@
<component name="InspectionProjectProfileManager">
<profile version="1.0">
<option name="myName" value="Project Default" />
</profile>
</component>

14
.idea/misc.xml 100644
View File

@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ExternalStorageConfigurationManager" enabled="true" />
<component name="MavenProjectsManager">
<option name="originalFiles">
<list>
<option value="$PROJECT_DIR$/pom.xml" />
</list>
</option>
</component>
<component name="ProjectRootManager" version="2" languageLevel="JDK_17" default="true" project-jdk-name="17" project-jdk-type="JavaSDK">
<output url="file://$PROJECT_DIR$/out" />
</component>
</project>

6
.idea/vcs.xml 100644
View File

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>

171
pom.xml 100644
View File

@ -0,0 +1,171 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.load</groupId>
<artifactId>Load_center</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<!-- 规定SpringBoot版本 -->
<parent>
<artifactId>spring-boot-starter-parent</artifactId>
<groupId>org.springframework.boot</groupId>
<version>2.6.2</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.5</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<!-- redis-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
<exclusions>
<exclusion>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.16.0</version>
</dependency>
<!-- http协议 -->
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpcore</artifactId>
<version>4.4.12</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.13</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.12.0</version>
</dependency>
<!-- web -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- 测试 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!-- Druid -->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid-spring-boot-starter</artifactId>
<version>1.2.8</version>
</dependency>
<!-- servlet 依赖. -->
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>8.0.11</version>
<scope>runtime</scope>
</dependency>
<!-- Mybatis 依赖配置 -->
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>2.2.2</version>
</dependency>
<!-- PageHelper mybatis分页插件 -->
<dependency>
<groupId>com.github.pagehelper</groupId>
<artifactId>pagehelper-spring-boot-starter</artifactId>
<version>1.4.1</version>
</dependency>
<!-- lombok小辣椒依赖 -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId>
<version>2.6</version>
</dependency>
<!-- json-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.79</version>
</dependency>
<!-- quartz定时任务-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-annotations</artifactId>
</dependency>
<!-- mq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>ecs20140526</artifactId>
<version>5.1.8</version>
</dependency>
</dependencies>
</project>

19
src/main/java/com/load/LoadApplication.java 100644
View File

@ -0,0 +1,19 @@
package com.load;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* 启动类
*
* @author YunFei·Du
* @ClassName: LoadApplication
* @Description: 启动类
* @CreateTime: 2024/5/27 09:50
*/
@SpringBootApplication
public class LoadApplication {
public static void main(String[] args) {
SpringApplication.run(LoadApplication.class,args);
}
}

10
src/main/java/com/load/MqttApplication.java 100644
View File

@ -0,0 +1,10 @@
package com.load;
/**
* @ClassName MqttApplication
* @Description 描述
* @Author YunFei.Du
* @Date 2024/5/30 9:19
*/
public class MqttApplication {
}

61
src/main/java/com/load/SubscribeSample.java 100644
View File

@ -0,0 +1,61 @@
package com.load;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
/**
* 订阅 MQTT 主题
*/
public class SubscribeSample {
public static void main(String[] args) {
/**
* 代理地址
*/
String broker = "tcp://39.100.87.192:1883";
/**
* 主题
*/
String topic = "mqtt/test";
String username = "emqx";
String password = "public";
/**
* 客户端ID(随机)
*/
String clientid = "subscribe_client";
int qos = 0;
try {
MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
// 连接参数
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(username);
options.setPassword(password.toCharArray());
options.setConnectionTimeout(60);
// 堵塞60S
options.setKeepAliveInterval(60);
// 设置回调
client.setCallback(new MqttCallback() {
public void connectionLost(Throwable cause) {
System.out.println("connectionLost: " + cause.getMessage());
}
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()));
}
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
});
client.connect(options);
client.subscribe(topic, qos);
} catch (Exception e) {
e.printStackTrace();
}
}
}

171
src/main/java/com/load/config/InitConnectWeight.java 100644
View File

@ -0,0 +1,171 @@
package com.load.config;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.ecs20140526.Client;
import com.aliyun.ecs20140526.models.DescribeInstancesRequest;
import com.aliyun.ecs20140526.models.DescribeInstancesResponse;
import com.aliyun.tea.TeaException;
import com.aliyun.teautil.models.RuntimeOptions;
import com.load.domain.ConnectWeight;
import com.load.util.ECSTool;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.*;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.stream.Collectors;
/**
* 初始化连接权重
*
* @author YunFei·Du
* @ClassName: InitConnectWeight
* @Description: 初始化连接权重
* @CreateTime: 2024/5/27 21:31
*/
@Component
@Slf4j
public class InitConnectWeight implements ApplicationRunner {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private RestTemplate restTemplate;
@Override
public void run(ApplicationArguments args) throws Exception {
ArrayList<ConnectWeight> connectWeightList = new ArrayList<>();
ArrayList<String> ipList = new ArrayList<>();
Client client = null;
//获取阿里云客户端
try {
client = ECSTool.createClient();
} catch (Exception e) {
throw new RuntimeException(e);
}
//获取所有实例
DescribeInstancesRequest describeInstancesRequest = new DescribeInstancesRequest()
.setRegionId("cn-zhangjiakou");
// .setInstanceType("ecs.e-c1m2.xlarge");
RuntimeOptions runtime = new RuntimeOptions();
try {
// 复制代码运行请自行打印 API 的返回值
DescribeInstancesResponse describeInstancesResponse = client.describeInstancesWithOptions(describeInstancesRequest, runtime);
List<List<String>> ipListList = describeInstancesResponse.getBody().instances.getInstance().stream().map(instance -> instance.publicIpAddress.ipAddress).collect(Collectors.toList());
for (List<String> strings : ipListList) {
for (String ip : strings) {
ipList.add(ip);
}
}
} catch (TeaException error) {
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
System.out.println(error.getMessage());
// 诊断地址
System.out.println(error.getData().get("Recommend"));
com.aliyun.teautil.Common.assertAsString(error.message);
} catch (Exception _error) {
TeaException error = new TeaException(_error.getMessage(), _error);
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
System.out.println(error.getMessage());
// 诊断地址
System.out.println(error.getData().get("Recommend"));
com.aliyun.teautil.Common.assertAsString(error.message);
}
//http://fluxmq.muyu.icu/public/login
//遍历所有ip,获取每一个服务的连接数
for (String ip : ipList) {
String url = "http://" + ip + ":8080/public/login";
Map<String, Object> request = new HashMap<>();
request.put("username", "fluxmq");
request.put("password", "fluxmq");
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<Map<String, Object>> r = new HttpEntity<Map<String, Object>>(request, httpHeaders);
String result = restTemplate.postForObject(url, r, String.class);
//http://fluxmq.muyu.icu/public/cluster
int nextInt = new Random().nextInt(1000);
String getInfoUrl = "http://" + ip + ":8080/public/cluster?random=" + nextInt;
HttpHeaders httpHeadersGetInfo = new HttpHeaders();
httpHeadersGetInfo.setContentType(MediaType.APPLICATION_JSON);
httpHeadersGetInfo.setAccept(Collections.singletonList(MediaType.APPLICATION_JSON));
httpHeadersGetInfo.set("Cookie", result);
HttpEntity getInfoRequest = new HttpEntity(httpHeadersGetInfo);
ResponseEntity<String> responseInfo = restTemplate.exchange(getInfoUrl, HttpMethod.GET, getInfoRequest, String.class, 1);
log.error("响应是:{}", responseInfo.getBody());
JSONArray jsonArray = JSON.parseArray(responseInfo.getBody());
if (jsonArray.size() > 0) {
JSONObject jsonObject = jsonArray.getJSONObject(0);
Integer connectSize = Integer.valueOf(jsonObject.getJSONObject("mqttInfo").getString("connectSize"));
connectWeightList.add(new ConnectWeight(ip,100-connectSize));
log.error("链接数量:{}", connectSize);
} else {
log.error("得到的相应数据为null");
}
}
// 初始化权重求和变量
Integer sum =0;
// 遍历ConnectWeight列表，计算所有权重值的总和
for (ConnectWeight connectWeight : connectWeightList) {
sum = sum + connectWeight.getWeightValue();
}
// 初始化最大轮询次数为0
int max = 0;
for (ConnectWeight connectWeight : connectWeightList) {
log.error("权重值:{}",connectWeight.getWeightValue());
// 将权重值转换为百分比并向下取整，得到每一轮的轮询次数
Integer result = BigDecimal.valueOf(connectWeight.getWeightValue() * 100).divide(BigDecimal.valueOf(sum), 0, RoundingMode.DOWN).intValue();
// 如果计算出的轮询次数大于当前最大值，更新最大值
if (result > max){
max = result;
}
// 更新ConnectWeight对象的权重值为计算出的轮询次数
connectWeight.setWeightValue(result);
// 输出100次轮询次数
log.error("100次轮询次数:{}",result);
}
ArrayList<String> weightIpList = new ArrayList<>();
//轮询出现次数
for (int i = 0; i <= max; i++) {
for (ConnectWeight connectWeight : connectWeightList) {
if (connectWeight.getWeightValue() > i){
weightIpList.add(connectWeight.getCarServerIp());
}else if (connectWeight.getWeightValue() == max){
weightIpList.add(connectWeight.getCarServerIp());
}
}
}
//存入redis
redisTemplate.delete("ipList");
for (String ip : weightIpList) {
redisTemplate.opsForList().rightPush("ipList",ip);
}
}
}

54
src/main/java/com/load/config/RabbitmqConfig.java 100644
View File

@ -0,0 +1,54 @@
package com.load.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* rabbitMq配置类
*
* @author YunFei·Du
* @ClassName: RabbitmqConfig
* @Description: rabbitMq配置类
* @CreateTime: 2024/5/27 16:56
*/
@Configuration
public class RabbitmqConfig {
public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
public static final String QUEUE_INFORM_SMS = "disconnect_connect";
public static final String EXCHANGE_TOPICS_INFORM="exchange_topics_inform";
public static final String ROUTINGKEY_EMAIL="inform.#.email.#";
public static final String ROUTINGKEY_SMS="inform.#.sms.#";
@Bean(EXCHANGE_TOPICS_INFORM)
public Exchange EXCHANGE_TOPICS_INFORM(){
//durable(true) 持久化，mq重启之后交换机还在
return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
}
//声明QUEUE_INFORM_EMAIL队列
@Bean(QUEUE_INFORM_EMAIL)
public Queue QUEUE_INFORM_EMAIL(){
return new Queue(QUEUE_INFORM_EMAIL);
}
//声明QUEUE_INFORM_SMS队列
@Bean(QUEUE_INFORM_SMS)
public Queue QUEUE_INFORM_SMS(){
return new Queue(QUEUE_INFORM_SMS);
}
//ROUTINGKEY_EMAIL队列绑定交换机，指定routingKey
@Bean
public Binding BINDING_QUEUE_INFORM_EMAIL(@Qualifier(QUEUE_INFORM_EMAIL) Queue queue,
@Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_EMAIL).noargs();
}
//ROUTINGKEY_SMS队列绑定交换机，指定routingKey
@Bean
public Binding BINDING_ROUTINGKEY_SMS(@Qualifier(QUEUE_INFORM_SMS) Queue queue,
@Qualifier(EXCHANGE_TOPICS_INFORM) Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_SMS).noargs();
}
}

33
src/main/java/com/load/config/RedisConfig.java 100644
View File

@ -0,0 +1,33 @@
package com.load.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
* redsi配置类
*
* @author YunFei·Du
* @ClassName: RedisConfoig
* @Description: redsi配置类
* @CreateTime: 2024/5/27 14:16
*/
@Configuration
public class RedisConfig {
@Bean
public RedisTemplate<String,String> redisTemplate(RedisConnectionFactory redisConnectionFactory){
RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(redisConnectionFactory);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer<Object>(Object.class));
redisTemplate.setHashKeySerializer(new StringRedisSerializer());
redisTemplate.setHashValueSerializer(new StringRedisSerializer());
return redisTemplate;
}
}

33
src/main/java/com/load/config/RestTemplateConfig.java 100644
View File

@ -0,0 +1,33 @@
package com.load.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.client.ClientHttpRequestFactory;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.web.client.RestTemplate;
/**
* rest配置类
*
* @author YunFei·Du
* @ClassName: RestTemplateConfig
* @Description: rest配置类
* @CreateTime: 2024/5/27 10:01
*/
@Configuration
public class RestTemplateConfig {
@Bean
public RestTemplate restTemplate(ClientHttpRequestFactory factory) {
return new RestTemplate(factory);
}
@Bean
public ClientHttpRequestFactory simpleClientHttpRequestFactory() {
SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
//超时设置
factory.setReadTimeout(5000);//ms
factory.setConnectTimeout(15000);//ms
return factory;
}
}

42
src/main/java/com/load/consumer/ReceiveHandler.java 100644
View File

@ -0,0 +1,42 @@
package com.load.consumer;
import com.load.config.InitConnectWeight;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.DefaultApplicationArguments;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
/**
* 断开连接消费者
*
* @author YunFei·Du
* @ClassName: ReceiveHandler
* @Description: 断开连接消费者
* @CreateTime: 2024/5/27 17:00
*/
@Component
@Slf4j
public class ReceiveHandler {
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private InitConnectWeight initConnectWeight;
//监听sms队列 ADD_LOG_AAA
@RabbitListener(queues = {"ADD_LOG_AAA"})
public void receiveSms(Message message) {
try {
initConnectWeight.run(new DefaultApplicationArguments());
redisTemplate.delete("count");
log.warn("ip权重重新分配成功!");
} catch (Exception e) {
throw new RuntimeException(e);
}
System.out.println("消费者得到的消息: " + new String(message.getBody()));
}
}

47
src/main/java/com/load/controller/LoadBalanceController.java 100644
View File

@ -0,0 +1,47 @@
package com.load.controller;
import com.load.domain.MqttServerModel;
import com.load.domain.Result;
import com.load.domain.req.VehicleConnectionReq;
import com.load.service.LoadBalanceService;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
/**
* 车辆连接
*
* @author YunFei·Du
* @ClassName: ConnectController
* @Description: 车辆连接
* @CreateTime: 2024/5/27 09:00
*/
@RestController
@RequestMapping("loadBalance")
@Log4j2
public class LoadBalanceController {
@Autowired
private LoadBalanceService loadBalanceService;
/**
* 获取连接
*/
@PostMapping("/getConnect")
public Result<MqttServerModel> getConnect(@RequestBody VehicleConnectionReq vehicleConnectionReq){
return loadBalanceService.getConnect(vehicleConnectionReq);
}
/**
* 创建实例
*/
@GetMapping("/createConnect")
public void createConnect() {
loadBalanceService.createConnect ( );
log.info ( "创建实例成功" );
}
}

32
src/main/java/com/load/domain/ConnectUseInfo.java 100644
View File

@ -0,0 +1,32 @@
package com.load.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 内存使用情况
*
* @author YunFei·Du
* @ClassName: MemoryUserInfo
* @Description: 内存使用情况
* @CreateTime: 2024/5/27 15:53
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ConnectUseInfo {
/**
* 节点ID
*/
private String clusterId;
/**
* 所属IP
*/
private String ipAddress;
/**
* 剩余可连接数量
*/
private Integer remainingNum;
}

28
src/main/java/com/load/domain/ConnectWeight.java 100644
View File

@ -0,0 +1,28 @@
package com.load.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 连接权重
*
* @author YunFei·Du
* @ClassName: CarServer
* @Description: 车子服务器
* @CreateTime: 2024/5/27 09:04
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class ConnectWeight {
/**
* 服务器ip
*/
private String carServerIp;
/**
* 权重值
*/
private Integer weightValue;
}

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

@ -0,0 +1,93 @@
package com.load.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/load/domain/MqttServerModel.java 100644
View File

@ -0,0 +1,28 @@
package com.load.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* Mqtt服务器模型
* @author YunFei.Du
* @date 22:08 2024/5/29
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MqttServerModel {
/**
* MQTT服务节点
*/
private String broker;
/**
* MQTT订阅主题
*/
private String topic;
}

101
src/main/java/com/load/domain/Result.java 100644
View File

@ -0,0 +1,101 @@
package com.load.domain;
import lombok.Data;
import java.io.Serializable;
/**
* 响应信息主体
*
* @author ruoyi
*/
@Data
public class Result<T> implements Serializable {
/**
* 成功
*/
public static final int SUCCESS = HttpStatus.SUCCESS;
/**
* 失败
*/
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;
private T data;
public static <T> Result<T> success () {
return restResult(null, SUCCESS, "操作成功");
}
public static <T> Result<T> success (T data) {
return restResult(data, SUCCESS, "操作成功");
}
public static <T> Result<T> success (T data, String msg) {
return restResult(data, SUCCESS, msg);
}
public static <T> Result<T> error () {
return restResult(null, FAIL, "操作失败");
}
public static <T> Result<T> error (String msg) {
return restResult(null, FAIL, msg);
}
public static <T> Result<T> error (T data) {
return restResult(data, FAIL, "操作失败");
}
public static <T> Result<T> error (T data, String msg) {
return restResult(data, FAIL, msg);
}
public static <T> Result<T> error (int code, String msg) {
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) {
Result<T> apiResult = new Result<>();
apiResult.setCode(code);
apiResult.setData(data);
apiResult.setMsg(msg);
return apiResult;
}
public static <T> Boolean isError (Result<T> ret) {
return !isSuccess(ret);
}
public static <T> Boolean isSuccess (Result<T> ret) {
return Result.SUCCESS == ret.getCode();
}
}

19
src/main/java/com/load/domain/VinIp.java 100644
View File

@ -0,0 +1,19 @@
package com.load.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* @ClassName VinIp
* @Description 描述
* @Author YunFei.Du
* @Date 2024/5/30 20:34
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class VinIp {
String vin;
String ip;
}

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

@ -0,0 +1,51 @@
package com.load.domain.req;
import com.alibaba.fastjson.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 vehicleVin;
/**
* 时间戳
*/
private String timestamp;
/**
* 用户名
*/
@JSONField(name = "username")
private String userName;
/**
* 随机字符串
*/
private String nonce;
}

32
src/main/java/com/load/mqtt/MessageCallbackService.java 100644
View File

@ -0,0 +1,32 @@
package com.load.mqtt;
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;
/**
* 消息回调服务 (回执消息类)
* @author YunFei.Du
* @date 22:37 2024/5/30
*/
@Service
public class MessageCallbackService 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());
}
}

46
src/main/java/com/load/mqtt/MqttFactory.java 100644
View File

@ -0,0 +1,46 @@
package com.load.mqtt;
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;
/**
* Mqtt工厂
* @author YunFei.Du
* @date 22:38 2024/5/30
*/
@Service
@AllArgsConstructor
public class MqttFactory {
private final MessageCallbackService messageCallbackService;
// 连接参数
/**
* 创建Mqtt 客户端
* @param mqttProperties
* @return
*/
public MqttClient createClient(MqttProperties mqttProperties){
MqttClient mqttClient =null;
try {
mqttClient=new MqttClient ( mqttProperties.getBroker() , mqttProperties.getClientId() , new MemoryPersistence() );
} catch (MqttException e) {
throw new RuntimeException ( e );
}
MqttConnectOptions options = new MqttConnectOptions ( );
if (mqttProperties.isLogin()){
options.setUserName ( mqttProperties.getUsername() );
options.setPassword ( mqttProperties.getPassword().toCharArray() );
}
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
mqttClient.setCallback ( messageCallbackService );
return mqttClient;
}
}

51
src/main/java/com/load/mqtt/MqttProperties.java 100644
View File

@ -0,0 +1,51 @@
package com.load.mqtt;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
/**
* 配置中心
* @author YunFei.Du
* @date 8:53 2024/5/30
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class MqttProperties {
private String broker;
private String topic;
private String username;
private String password;
private String clientId;
/**
* 构建mqtt配置
* @param ip
* @param topic
* @return
*/
public static MqttProperties configBuild(String ip, String topic){
return MqttProperties.builder()
.broker("tcp://"+ip+":1883")
.topic(topic)
.username("admin")
.password("public")
.clientId("protocol-parsing") //协议解析 定值 --> 配置
.build();
}
/**
* 判断是否可以登录
* @return
*/
public boolean isLogin(){
// commons-lang3
return StringUtils.isBlank ( username ) && !StringUtils.isBlank ( password );
}
}

28
src/main/java/com/load/rebbitmq/MsgHandle.java 100644
View File

@ -0,0 +1,28 @@
package com.load.rebbitmq;
import com.load.mqtt.MqttFactory;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 信息处理器
* @author YunFei.Du
* @date 9:21 2024/5/30
*/
@Component
@Log4j2
public class MsgHandle {
@Autowired
private MqttFactory mqttFactory;
// @RabbitListener(queues = {"create.topic"})
// private void msg(String msg){
// log.info ( "接收到消息：{}" , msg );
// MqttProperties mqttProperties = MqttProperties.configBuild ( "39.100.87.192", "mqtt/test" );
// mqttFactory.createClient ( mqttProperties );
//
// }
}

36
src/main/java/com/load/rebbitmq/RabbitConfig.java 100644
View File

@ -0,0 +1,36 @@
package com.load.rebbitmq;
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;
/**
* rabbit 配置类
* @author YunFei.Du
* @date 8:39 2024/5/31
*/
@Configuration
public class RabbitConfig {
@Bean
public Queue initQueue() {
return new Queue ("create.totic", true);
}
@Bean
public DirectExchange direct(){
return new DirectExchange ("topic.direct");
}
@Bean
public Binding bindingla(DirectExchange direct, Queue initQueue){
return BindingBuilder.bind(initQueue)
.to(direct)
.with("protocol-parsing");
}
}

27
src/main/java/com/load/service/LoadBalanceService.java 100644
View File

@ -0,0 +1,27 @@
package com.load.service;
/**
* 连接服务接口
*
* @author YunFei·Du
* @ClassName: ConnectServer
* @Description: 连接服务接口
* @CreateTime: 2024/5/27 09:22
*/
import com.load.domain.MqttServerModel;
import com.load.domain.Result;
import com.load.domain.req.VehicleConnectionReq;
/**
*@ClassName ConnectServer
*@Description 描述
*@Author YunFei·Du
*@Date 2024/5/27 09:22
*/
public interface LoadBalanceService {
Result<MqttServerModel> getConnect(VehicleConnectionReq vehicleConnectionReq);
void createConnect();
}

81
src/main/java/com/load/service/impl/LoadBalanceServiceImpl.java 100644
View File

@ -0,0 +1,81 @@
package com.load.service.impl;
import com.load.domain.MqttServerModel;
import com.load.domain.Result;
import com.load.domain.VinIp;
import com.load.domain.req.VehicleConnectionReq;
import com.load.service.LoadBalanceService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import static com.load.util.ECSTool.runEcsInstance;
/**
* 服务实现类
*
* @author YunFei·Du
* @ClassName: ConnectServiceImpl
* @Description: 服务实现类
* @CreateTime: 2024/5/27 09:22
*/
@Service
@Slf4j
public class LoadBalanceServiceImpl implements LoadBalanceService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Override
public Result<MqttServerModel> getConnect(VehicleConnectionReq vehicleConnectionReq) {
log.error(vehicleConnectionReq.toString());
//判断是否有游标key --count
if (redisTemplate.hasKey("count")){
Integer count = Integer.valueOf(redisTemplate.opsForValue().get("count"));
if (count == 100){
redisTemplate.opsForValue().set("count",String.valueOf(0));
}else {
redisTemplate.opsForValue().set("count",String.valueOf(count + 1));
}
//根据游标count获取服务IP
String ip = redisTemplate.opsForList().index("ipList", count);
//关联车辆和服务
this.insertVinIp(new VinIp (vehicleConnectionReq.getVehicleVin(),ip));
//响应信息
return Result.success(new MqttServerModel("tcp://"+ip+":1883","test1"));
}else {
redisTemplate.opsForValue().set("count",String.valueOf(1));
//根据游标count获取服务IP
String ip = redisTemplate.opsForList().index("ipList", 0);
//关联车辆和服务
this.insertVinIp(new VinIp(vehicleConnectionReq.getVehicleVin(),ip));
//响应信息
return Result.success(new MqttServerModel("tcp://"+ip+":1883","test1"));
}
}
private void insertVinIp(VinIp vinIp) {
}
@SneakyThrows
@Override
public void createConnect() {
runEcsInstance("cn-zhangjiakou", "lt-8vbepqjihmawbkqcwkcm");
}
}

150
src/main/java/com/load/util/ECSTool.java 100644
View File

@ -0,0 +1,150 @@
package com.load.util;
import com.aliyun.ecs20140526.Client;
import com.aliyun.ecs20140526.models.*;
import com.aliyun.tea.TeaException;
import com.aliyun.teaopenapi.models.Config;
import com.aliyun.teautil.models.RuntimeOptions;
import lombok.extern.log4j.Log4j2;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
/**
* ECS实例工具类
* @author YunFei.Du
* @date 9:30 2024/5/28
*/
@Log4j2
public class ECSTool {
/**
* 创建ECS客户端
* @return ECS客户端实例
* @throws Exception 如果配置信息不正确或网络问题，将抛出异常
* 注意：此方法用于演示，实际使用时请替换为安全的鉴权方式，如STS
* 更多鉴权访问方式参考：https://help.aliyun.com/document_detail/378657.html
*/
public static Client createClient() throws Exception {
// 工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考。
Config config = new Config()
.setAccessKeyId("LTAI5tPTk3MFkmCGBbnQgmrM")
.setAccessKeySecret("q7rLjxrI0SLBXlvNT4VmYcHCNCY2p6");
// Endpoint 请参考 https://api.aliyun.com/product/Ecs
config.endpoint = "ecs.cn-zhangjiakou.aliyuncs.com";
return new Client(config);
}
/**
* 创建并运行ECS实例
* @param regionId 地域ID
* @param launchTemplateId 启动模板ID
* @throws Exception 如果调用API时发生错误，将抛出异常
*/
public static void runEcsInstance(String regionId, String launchTemplateId) throws Exception {
// 创建ECS客户端
Client client = ECSTool.createClient();
// 设置运行实例的请求参数
RunInstancesRequest runInstancesRequest = new RunInstancesRequest()
.setRegionId(regionId)
.setLaunchTemplateId(launchTemplateId);
// 创建运行选项
RuntimeOptions runtime = new RuntimeOptions();
try {
// 调用API运行实例
client.runInstancesWithOptions(runInstancesRequest, runtime);
} catch (Exception error) {
// 处理API调用过程中出现的异常
System.out.println(error.getMessage());
if (error instanceof TeaException) {
// 处理特定类型的异常，如TeaException
TeaException teaError = (TeaException) error;
System.out.println(teaError.getData().get("Recommend")); // 打印诊断推荐链接
com.aliyun.teautil.Common.assertAsString(teaError.getMessage()); // 断言错误信息
} else {
// 处理其他类型的异常
System.out.println(error.getMessage());
}
}
}
/**
* 销毁实例
*/
public static void runEcsRemove(String instanceId) throws Exception {
// 创建ECS客户端
Client client = ECSTool.createClient();
DeleteInstanceRequest deleteInstanceRequest = new DeleteInstanceRequest ()
.setInstanceId(instanceId);
RuntimeOptions runtime = new RuntimeOptions();
try {
// 复制代码运行请自行打印 API 的返回值
client.deleteInstanceWithOptions(deleteInstanceRequest, runtime);
} catch (TeaException error) {
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
System.out.println(error.getMessage());
// 诊断地址
System.out.println(error.getData().get("Recommend"));
com.aliyun.teautil.Common.assertAsString(error.message);
} catch (Exception _error) {
TeaException error = new TeaException(_error.getMessage(), _error);
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
System.out.println(error.getMessage());
// 诊断地址
System.out.println(error.getData().get("Recommend"));
com.aliyun.teautil.Common.assertAsString(error.message);
}
}
/**
* 查询实例列表
* @param regionId
* @return
* @throws Exception
*/
public static List<String> FindInstance( String regionId) throws Exception {
// 创建ECS客户端
Client client = ECSTool.createClient ( );
DescribeInstancesRequest describeInstancesRequest = new DescribeInstancesRequest ( )
.setRegionId ( regionId );
RuntimeOptions runtime = new RuntimeOptions ( );
List< String > ipList = new ArrayList<> ( );
try {
// 复制代码运行请自行打印 API 的返回值
DescribeInstancesResponse response = client.describeInstancesWithOptions ( describeInstancesRequest, runtime );
List< List< String > > ipListList = response.getBody ( ).instances.getInstance ( ).stream ( ).map ( instance -> instance.publicIpAddress.ipAddress ).collect ( Collectors.toList ( ) );
for (List< String > strings : ipListList) {
for (String ip : strings) {
ipList.add ( ip );
}
System.out.println ( "------------------------" );
}
log.info ( "ipList: " + ipList ); // [39.100.89.218, 39.100.87.192]
return ipList;
} catch (TeaException error) {
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
System.out.println ( error.getMessage ( ) );
// 诊断地址
System.out.println ( error.getData ( ).get ( "Recommend" ) );
com.aliyun.teautil.Common.assertAsString ( error.message );
} catch (Exception _error) {
TeaException error = new TeaException ( _error.getMessage ( ), _error );
// 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
// 错误 message
System.out.println ( error.getMessage ( ) );
// 诊断地址
System.out.println ( error.getData ( ).get ( "Recommend" ) );
com.aliyun.teautil.Common.assertAsString ( error.message );
}
return null;
}
}

24
src/main/resources/application.yml 100644
View File

@ -0,0 +1,24 @@
# 服务器相关
server:
port: 82
spring:
application:
# 协议解析
name: protocol-parsing
rabbitmq:
host: 111.229.102.61
port: 5672
username: guest
password: guest
virtualHost: /
redis:
host: 127.0.0.1
port: 6379
password: dyf@123
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://111.229.102.61/car?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&useSSL=false
username: root
password: dyf@123