five-groups
/
five-groups-couplet
4
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

新增

server_five_liuyunhu
lijiayao 2024-04-02 21:25:33 +08:00
parent aaa5d4f868
commit a9977b44a9
34 changed files with 870 additions and 238 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-remote/pom.xml
View File

@ -1,26 +0,0 @@
<?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>
<parent>
<groupId>com.couplet</groupId>
<artifactId>couplet-analyze-incident</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>couplet-incident-remote</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.couplet</groupId>
<artifactId>couplet-incident-remote</artifactId>
</dependency>
</dependencies>
</project>

20
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/pom.xml
View File

@ -1,20 +0,0 @@
<?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>
<parent>
<groupId>com.couplet</groupId>
<artifactId>couplet-analyze-incident</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>couplet-incident-server</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
</project>

19
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/java/com/couplet/analyze/incident/server/controller/IncidentController.java
View File

@ -1,19 +0,0 @@
package com.couplet.analyze.incident.server.controller;
import com.couplet.common.core.web.controller.BaseController;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* 事件控制器
*/
@RestController
@RequestMapping("/incident")
public class IncidentController extends BaseController {
}

13
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/java/com/couplet/analyze/incident/server/service/IncidentService.java
View File

@ -1,13 +0,0 @@
package com.couplet.analyze.incident.server.service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 事件系统接口
*/
public interface IncidentService {
}

13
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/java/com/couplet/analyze/incident/server/service/impl/BreakdownServiceImpl.java
View File

@ -1,13 +0,0 @@
package com.couplet.analyze.incident.server.service.impl;
import com.couplet.analyze.incident.server.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 故障事件
*/
@Service("breakdown")
public class BreakdownServiceImpl implements IncidentService {
}

17
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/java/com/couplet/analyze/incident/server/service/impl/ElectronicFenceServiceImpl.java
View File

@ -1,17 +0,0 @@
package com.couplet.analyze.incident.server.service.impl;
import com.couplet.analyze.incident.server.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 电子围栏事件服务实现类
*/
@Service("electronic-fence")
public class ElectronicFenceServiceImpl implements IncidentService {
}

13
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/java/com/couplet/analyze/incident/server/service/impl/RealTimeDataServiceImpl.java
View File

@ -1,13 +0,0 @@
package com.couplet.analyze.incident.server.service.impl;
import com.couplet.analyze.incident.server.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 实时数据事件
*/
@Service("real-time-data")
public class RealTimeDataServiceImpl implements IncidentService {
}

13
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/java/com/couplet/analyze/incident/server/service/impl/StoredEventServiceImpl.java
View File

@ -1,13 +0,0 @@
package com.couplet.analyze.incident.server.service.impl;
import com.couplet.analyze.incident.server.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 事件存储服务
*/
@Service("stored-event")
public class StoredEventServiceImpl implements IncidentService {
}

31
couplet-modules/couplet-analyze/couplet-analyze-incident/couplet-incident-server/src/main/resources/bootstrap.yml
View File

@ -1,31 +0,0 @@
# Tomcat
server:
port: 9999
# Spring
spring:
application:
# 应用名称
name: couplet-incident
profiles:
# 环境配置
active: dev
cloud:
nacos:
discovery:
# 服务注册地址
server-addr: 121.89.211.230:8848
config:
# 配置中心地址
server-addr: 121.89.211.230:8848
# 配置文件格式
file-extension: yml
# 共享配置
shared-configs:
- application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
main:
allow-bean-definition-overriding: true
logging:
level:
com.couplet.trouble.mapper: DEBUG

25
couplet-modules/couplet-analyze/couplet-analyze-incident/pom.xml
View File

@ -1,25 +0,0 @@
<?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>
<parent>
<groupId>com.couplet</groupId>
<artifactId>couplet-analyze</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>couplet-analyze-incident</artifactId>
<packaging>pom</packaging>
<modules>
<module>couplet-incident-remote</module>
<module>couplet-incident-server</module>
</modules>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
</project>

1
couplet-modules/couplet-analyze/couplet-analyze-msg/pom.xml
View File

@ -98,6 +98,7 @@
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
</dependencies>
</project>

49
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/config/RabbitAdminConfig.java 100644
View File

@ -0,0 +1,49 @@
package com.couplet.analyze.msg.config;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitAdminConfig {
@Value("${spring.rabbitmq.host}")
private String host;
@Value("${spring.rabbitmq.username}")
private String username;
@Value("${spring.rabbitmq.password}")
private String password;
@Value("${spring.rabbitmq.virtualhost}")
private String virtualhost;
/**
* 构建RabbitMQ的连接工厂
* @return
*/
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses(host);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualhost);
return connectionFactory;
}
/**
* 初始化RabbitAdmin
* @param connectionFactory
* @return
*/
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;
}
}

96
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/consumer/IncidentConsumer.java 100644
View File

@ -0,0 +1,96 @@
package com.couplet.analyze.msg.consumer;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import com.couplet.common.core.utils.SpringUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 消费传输过来的数据
*/
@Component
@Log4j2
@RabbitListener(queues = "couplet-msg-queue")
public class IncidentConsumer {
/**
* 调用事件服务接口
*/
@Autowired
private IncidentService incidentService;
@Autowired
private StringRedisTemplate redis;
static ArrayList<String> strings = new ArrayList<>(){
{
add("breakdown");
add("electronic-fence");
add("real-time-data");
add("stored-event");
}
};
@RabbitHandler
public void incidentConsumer(List<CoupletMsgData> list, Channel channel, Message message) throws IOException {
log.info("消费者收到的消息是:【{}】",list);
long deliveryTag = message.getMessageProperties().getDeliveryTag();
String messageId = message.getMessageProperties().getMessageId();
if (redis.hasKey("解析系统value:"+messageId)){
redis.opsForValue().set("解析系统value:"+messageId,deliveryTag+"",5, TimeUnit.MINUTES);
}
Long add = redis.opsForSet().add("解析系统:set" + messageId,messageId);
redis.expire("解析系统:set"+messageId,5,TimeUnit.MINUTES);
try {
if (0 < add){
log.info("消费者开始消费，消费者收到消息是:【{ }】"+list);
for (CoupletMsgData coupletMsgData : list) {
for (String string : strings) {
CoupletMsgData bean = SpringUtils.getBean(string);
incidentService.incident(coupletMsgData);
}
}
channel.basicAck(deliveryTag,false);
} else {
//重复消费
log.error("重复消费");
//拒绝消费
channel.basicReject(deliveryTag, false);
//删除缓存
redis.opsForSet().remove("解析系统:set"+messageId, messageId);
}
} catch (Exception e) {
log.error("消息消费失败...消息信息为:【{ }】", list);
String s = redis.opsForValue().get("解析系统value:" + messageId);
Long o = Long.parseLong( s );
if (deliveryTag == o+2){
log.error("队列无法进入了,消息是:【{}】",list);
channel.basicNack(deliveryTag,false,false);
} else {
log.info("再一次进入队列，消息是:[{}]", list);
channel.basicNack(deliveryTag,false,true);
}
}
}
}

47
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/controller/IncidentController.java 100644
View File

@ -0,0 +1,47 @@
package com.couplet.analyze.msg.controller;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import com.couplet.common.core.domain.Result;
import com.couplet.common.core.web.controller.BaseController;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.ArrayList;
import java.util.List;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* 事件控制器
*/
@RestController
@RequestMapping("/incident")
public class IncidentController extends BaseController {
/**
* 事件列表
*/
static List<String> list=new ArrayList<>(){
{
add("breakdown");
add("electronic-fence");
add("real-time-data");
add("stored-event");
}
};
/**
* 事件服务
*/
@Autowired
private IncidentService incidentService;
}

4
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/domain/CoupletMsgData.java
View File

@ -260,4 +260,8 @@ public class CoupletMsgData {
* CHG(充电机)状态 1：正常 0：故障
*/
private int chgStatus;
public CoupletMsgData(String s) {
this.vin = s;
}
}

14
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/mapper/IncidentMapper.java 100644
View File

@ -0,0 +1,14 @@
package com.couplet.analyze.msg.mapper;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import org.apache.ibatis.annotations.Mapper;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description:
*/
@Mapper
public interface IncidentMapper {
public void reportMapper(CoupletMsgData coupletMsgData);
}

28
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/IncidentService.java 100644
View File

@ -0,0 +1,28 @@
package com.couplet.analyze.msg.service;
import com.couplet.analyze.msg.contents.MsgContent;
import com.couplet.analyze.msg.domain.CoupletMsgData;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 事件系统接口
*/
public interface IncidentService {
/**
* 事件分析
*
* @param coupletMsgData
*/
void incident(CoupletMsgData coupletMsgData);
/**
* 获取事件名称
* @return 事件名称
*/
String getName();
}

31
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/BreakdownServiceImpl.java 100644
View File

@ -0,0 +1,31 @@
package com.couplet.analyze.msg.service.impl;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 故障事件
*/
@Service("breakdown")
public class BreakdownServiceImpl implements IncidentService {
/**
* 故障事件
*
* @param coupletMsgData
*/
@Override
public void incident(CoupletMsgData coupletMsgData) {
}
/**
* @return 获取事件名称
*/
@Override
public String getName() {
return "breakdown";
}
}

33
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/ElectronicFenceServiceImpl.java 100644
View File

@ -0,0 +1,33 @@
package com.couplet.analyze.msg.service.impl;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 电子围栏事件服务实现类
*/
@Service("electronic-fence")
public class ElectronicFenceServiceImpl implements IncidentService {
/**
* 电子围栏事件
*
* @param coupletMsgData
*/
@Override
public void incident(CoupletMsgData coupletMsgData) {
}
/**
* @return 电子围栏service 名称
*/
@Override
public String getName() {
return "electronic-fence";
}
}

33
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/RealTimeDataServiceImpl.java 100644
View File

@ -0,0 +1,33 @@
package com.couplet.analyze.msg.service.impl;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 实时数据事件
*/
@Service("real-time-data")
public class RealTimeDataServiceImpl implements IncidentService {
/**
* 实时数据事件
*
* @param coupletMsgData
*/
@Override
public void incident(CoupletMsgData coupletMsgData) {
}
/**
* 实时数据事件
* @return
*/
@Override
public String getName() {
return "real-time-data";
}
}

41
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/StoredEventServiceImpl.java 100644
View File

@ -0,0 +1,41 @@
package com.couplet.analyze.msg.service.impl;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.mapper.IncidentMapper;
import com.couplet.analyze.msg.service.IncidentService;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description: 事件存储服务
*/
@Service("stored-event")
@Log4j2
public class StoredEventServiceImpl implements IncidentService {
@Autowired
private IncidentMapper incidentMapper;
/**
* 事件存储服务
*
* @param coupletMsgData
*/
@Override
public void incident(CoupletMsgData coupletMsgData) {
log.info("开始存储......");
incidentMapper.reportMapper(coupletMsgData);
log.info("结束存储......");
}
/**
* 事件存储服务
* @return
*/
@Override
public String getName() {
return "stored-event";
}
}

74
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/DLXQueueUtil.java 100644
View File

@ -0,0 +1,74 @@
package com.couplet.analyze.msg.utils;
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.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
@Component
public class DLXQueueUtil {
// routingKey
private static final String DEAD_ROUTING_KEY = "dead.routingkey";
private static final String ROUTING_KEY = "routingkey";
private static final String DEAD_EXCHANGE = "dead.exchange";
private static final String EXCHANGE = "common.exchange";
@Autowired
RabbitTemplate rabbitTemplate;
@Resource
RabbitAdmin rabbitAdmin;
/**
* 发送死信队列，过期后进入死信交换机，进入死信队列
* @param queueName 队列名称
* @param deadQueueName 死信队列名称
* @param params 消息内容
* @param expiration 过期时间 毫秒
*/
public void sendDLXQueue(String queueName, String deadQueueName, Object params, Integer expiration){
/**
* ----------------------------------先创建一个ttl队列和死信队列--------------------------------------------
*/
Map<String, Object> map = new HashMap<>();
// 队列设置存活时间，单位ms, 必须是整形数据。
map.put("x-message-ttl",expiration);
// 设置死信交换机
map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 设置死信交换器路由
map.put("x-dead-letter-routing-key", DEAD_ROUTING_KEY);
/*参数1：队列名称  参数2：持久化  参数3：是否排他 参数4：自动删除队列  参数5：队列参数*/
Queue queue = new Queue(queueName, true, false, false, map);
rabbitAdmin.declareQueue(queue);
/**
* ---------------------------------创建交换机---------------------------------------------
*/
DirectExchange directExchange = new DirectExchange(EXCHANGE, true, false);
rabbitAdmin.declareExchange(directExchange);
/**
* ---------------------------------队列绑定交换机---------------------------------------------
*/
Binding binding = BindingBuilder.bind(queue).to(directExchange).with(ROUTING_KEY);
rabbitAdmin.declareBinding(binding);
/**
* ---------------------------------在创建一个死信交换机和队列，接收死信队列---------------------------------------------
*/
DirectExchange deadExchange = new DirectExchange(DEAD_EXCHANGE, true, false);
rabbitAdmin.declareExchange(deadExchange);
Queue deadQueue = new Queue(deadQueueName,true,false,false);
rabbitAdmin.declareQueue(deadQueue);
/**
* ---------------------------------队列绑定死信交换机---------------------------------------------
*/
// 将队列和交换机绑定
Binding deadbinding = BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY);
rabbitAdmin.declareBinding(deadbinding);
// 发送消息
rabbitTemplate.convertAndSend(EXCHANGE,ROUTING_KEY,params);
}
}

67
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/DelayedQueueUtil.java 100644
View File

@ -0,0 +1,67 @@
package com.couplet.analyze.msg.utils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
@Component
public class DelayedQueueUtil {
// routingKey
private static final String DELAYED_ROUTING_KEY = "delayed.routingkey";
// 延迟队列交换机
private static final String DELAYED_EXCHANGE = "delayed.exchange";
@Autowired
RabbitTemplate rabbitTemplate;
@Resource
RabbitAdmin rabbitAdmin;
/**
* 发送延迟队列
*
* @param queueName 队列名称
* @param params 消息内容
* @param expiration 延迟时间 毫秒
*/
public void sendDelayedQueue(String queueName, Object params, Integer expiration) {
// 先创建一个队列
Queue queue = new Queue(queueName);
rabbitAdmin.declareQueue(queue);
// 创建延迟队列交换机
CustomExchange customExchange = createCustomExchange();
rabbitAdmin.declareExchange(customExchange);
// 将队列和交换机绑定
Binding binding = BindingBuilder.bind(queue).to(customExchange).with(DELAYED_ROUTING_KEY).noargs();
rabbitAdmin.declareBinding(binding);
// 发送延迟消息
rabbitTemplate.convertAndSend(DELAYED_EXCHANGE, DELAYED_ROUTING_KEY, params, message -> {
message.getMessageProperties().setDelay(expiration);
return message;
});
}
public CustomExchange createCustomExchange() {
Map<String, Object> arguments = new HashMap<>();
/**
* 参数说明：
* 1.交换机的名称
* 2.交换机的类型
* 3.是否需要持久化
* 4.是否自动删除
* 5.其它参数
*/
arguments.put("x-delayed-type", "direct");
return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", true, false, arguments);
}
}

118
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/MqttMonitor.java 100644
View File

@ -0,0 +1,118 @@
package com.couplet.analyze.msg.utils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* @ProjectName: five-groups-couplet
* @Author: LiuYunHu
* @CreateTime: 2024/4/1
* @Description: mqtt监听者
*/
@Component
@Slf4j
public class MqttMonitor {
/*
* 路径
* */
@Value("${mqtt.server.broker}")
private String broker;
/*
* 主题
* */
@Value("${mqtt.server.topic}")
private String topic;
/*
* 客户端id
* */
@Value("${mqtt.server.clientId}")
private String clientId;
/*
* 用户名
* */
@Value("${mqtt.server.username}")
private String userName;
/*
* 密码
* */
@Value("${mqtt.server.password}")
private String password;
/*
* 遗嘱消息qos
* */
@Value("${mqtt.server.qos}")
private Integer qos;
//随项目启动而执行这个方法
@PostConstruct
public void connect() {
log.info("mqtt监听者启动");
try {
MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(userName);
options.setPassword(password.toCharArray());
//连接超时
options.setConnectionTimeout(10);
//心跳时间
options.setKeepAliveInterval(60);
log.info("mqtt监听者启动成功，连接到：{}", broker);
client.connect(options);
log.info("连接成功！");
//设置回调
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable throwable) {
log.error("连接丢失：{}", throwable.getMessage());
}
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) {
// log.info("消息已送达");
// log.info("接收消息主题：{}",topic);
// log.info("接收消息qos：{}", mqttMessage.getQos());
//接收到的原始报文
String message = new String(mqttMessage.getPayload());
log.info("接收消息原始内容：{}", message);
//去除空格 得到16进制字符串
String replaced = message.replaceAll(" ", "");
log.info("接收消息剪切后内容：{}", replaced);
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("消息发送成功！");
}
});
client.subscribe(topic, qos);
} catch (MqttException e) {
log.error("mqtt监听者启动失败，{}", e.getMessage());
throw new RuntimeException(e);
}
}
}

50
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/TTLQueueUtil.java 100644
View File

@ -0,0 +1,50 @@
package com.couplet.analyze.msg.utils;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
@Component
public class TTLQueueUtil {
// routingKey
private static final String TTL_KEY = "ttl.routingkey";
private static final String TTL_EXCHANGE = "ttl.exchange";
@Autowired
RabbitTemplate rabbitTemplate;
@Resource
RabbitAdmin rabbitAdmin;
/**
* 发送TTL队列
* @param queueName 队列名称
* @param params 消息内容
* @param expiration 过期时间 毫秒
*/
public void sendTtlQueue(String queueName, Object params, Integer expiration) {
/**
* ----------------------------------先创建一个ttl队列--------------------------------------------
*/
Map<String, Object> map = new HashMap<>();
// 队列设置存活时间，单位ms,必须是整形数据。
map.put("x-message-ttl",expiration);
/*参数1：队列名称  参数2：持久化  参数3：是否排他 参数4：自动删除队列  参数5：队列参数*/
Queue queue = new Queue(queueName,true,false,false,map);
rabbitAdmin.declareQueue(queue);
/**
* ---------------------------------创建交换机---------------------------------------------
*/
DirectExchange directExchange = new DirectExchange(TTL_EXCHANGE, true, false);
rabbitAdmin.declareExchange(directExchange);
/**
* ---------------------------------队列绑定交换机---------------------------------------------
*/
// 将队列和交换机绑定
Binding binding = BindingBuilder.bind(queue).to(directExchange).with(TTL_KEY);
rabbitAdmin.declareBinding(binding);
// 发送消息
rabbitTemplate.convertAndSend(TTL_EXCHANGE,TTL_KEY,params);
}
}

2
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml
View File

@ -29,7 +29,7 @@ spring:
allow-bean-definition-overriding: true
logging:
level:
com.couplet.msg.mapper: DEBUG
com.couplet.analyze.msg.mapper: DEBUG
mybatis-plus:
configuration:
map-underscore-to-camel-case: true

75
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/mapper/incident/SysIncidentMapper.xml 100644
View File

@ -0,0 +1,75 @@
<?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.couplet.analyze.msg.mapper.IncidentMapper">
<insert id="reportMapper">
INSERT INTO `vehicle-resolver`.`resolver_report_data`
(`vin`, `create_time`, `longitude`, `latitude`,
`speed`, `mileage`, `voltage`, `current`, `resistance`, `gear`,
`acceleration_pedal`, `fuel_consumption_rate`,
`motor_controller_temperature`, `motor_speed`,
`motor_torque`, `motor_temperature`, `motor_voltage`,
`motor_current`, `remaining_battery`, `maximum_feedback_power`,
`maximum_discharge_power`, `self_check_counter`,
`total_battery_current`, `total_battery_voltage`,
`single_battery_max_voltage`, `single_battery_min_voltage`,
`single_battery_max_temperature`, `single_battery_min_temperature`,
`available_battery_capacity`, `vehicle_status`, `charging_status`,
`operatingStatus`, `soc_status`, `charging_energy_storage_status`,
`drive_motor_status`, `position_status`, `eas_status`, `ptc_status`,
`eps_status`, `abs_status`, `mcu_status`, `heating_status`, `battery_status`,
`battery_insulation_status`, `dcdc_status`, `chg_status`, `brake_pedal`)
VALUES (#{vin},
#{createTime},
#{longitude},
#{latitude},
#{speed},
#{mileage},
#{voltage},
#{current},
#{resistance},
#{gear},
#{accelerationPedal},
#{fuelConsumptionRate},
#{motorControllerTemperature},
#{motorSpeed},
#{motorTorque},
#{motorTemperature},
#{motorVoltage},
#{motorCurrent},
#{remainingBattery},
#{maximumFeedbackPower},
#{maximumDischargePower},
#{selfCheckCounter},
#{totalBatteryCurrent},
#{totalBatteryVoltage},
#{singleBatteryMaxVoltage},
#{singleBatteryMinVoltage},
#{singleBatteryMaxTemperature},
#{singleBatteryMinTemperature},
#{availableBatteryCapacity},
#{vehicleStatus},
#{chargingStatus},
#{operatingStatus},
#{socStatus},
#{chargingEnergyStorageStatus},
#{driveMotorStatus},
#{positionStatus},
#{easStatus},
#{ptcStatus},
#{epsStatus},
#{absStatus},
#{mcuStatus},
#{heatingStatus},
#{batteryStatus},
#{batteryInsulationStatus},
#{dcdcStatus},
#{chgStatus},
#{brakePedal}
);
</insert>
</mapper>

41
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/mapper/trouble/SysTroubleMapper.xml
View File

@ -1,41 +0,0 @@
<?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.couplet.trouble.mapper.SysTroubleMapper">
<resultMap type="com.couplet.trouble.domain.CoupletTroubleCode" id="SysTroubleResult">
<id property="troubleId" column="trouble_id"/>
<result property="troubleCode" column="trouble_code"/>
<result property="troubleValue" column="trouble_value"/>
<result property="troublePosition" column="trouble_position"/>
<result property="troubleTag" column="trouble_tag"/>
<result property="typeId" column="type_id"/>
<result property="gradeId" column="grade_id"/>
</resultMap>
<sql id="selectTroubleVo">
select t.*,g.grade_name,y.type_name from couplet_trouble_code t
LEFT JOIN couplet_trouble_grade g on t.grade_id = g.grade_id
LEFT JOIN couplet_trouble_type y on t.type_id= y.type_id
</sql>
<select id="selectTroubleList" parameterType="com.couplet.trouble.mapper.SysTroubleMapper" resultMap="SysTroubleResult">
<include refid="selectTroubleVo"/>
<where>
<if test="troubleCode != null and troubleCode != ''">
AND trouble_code = #{troubleCode}
</if>
<if test="troublePosition != null and troublePosition != ''">
AND trouble_position like concat('%', #{troublePosition}, '%')
</if>
</where>
</select>
<select id="selectTroubleListByType" resultType="com.couplet.trouble.domain.CoupletTroubleType">
select * from couplet_trouble_type
</select>
<select id="selectTroubleListByGrade" resultType="com.couplet.trouble.domain.CoupletTroubleGrade">
select * from couplet_trouble_grade
</select>
</mapper>

92
couplet-modules/couplet-analyze/couplet-analyze-msg/src/test/java/com/couplet/msg/dome.java 100644
View File

@ -0,0 +1,92 @@
package com.couplet.msg;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import com.couplet.common.core.utils.SpringUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
* @Description:
*/
public class dome {
private static final List<String> msgList = new ArrayList<>(){{
add("7E 56 49 4e 31 32 33 34 35 36 37 38 39 44 49 4a 45 34 31 37 31 31 37 36 34 31 30 34 35 30 36 31 31 36 2e 36 36 34 33 38 30 30 33 39 2e 35 33 31 39 39 30 30 37 32 2e 30 30 30 33 31 2e 33 37 36 30 30 30 30 30 32 32 30 30 30 30 32 32 30 30 30 38 35 32 30 30 30 30 30 30 44 30 30 38 30 39 2e 36 30 30 39 34 30 30 30 30 35 38 39 30 36 36 37 39 30 39 33 30 30 30 30 32 30 33 30 30 32 30 33 30 30 30 30 30 34 34 32 38 32 2e 35 35 30 30 30 30 31 34 30 30 30 30 38 30 37 30 30 30 30 37 34 34 30 30 30 33 30 30 30 34 30 30 30 39 35 30 30 30 30 35 38 30 30 30 30 35 34 30 30 30 30 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 31 24 7E");
}};
/**
* 事件列表
*/
static List<String> list=new ArrayList<>(){
{
add("breakdown");
add("electronic-fence");
add("real-time-data");
add("stored-event");
}
};
static IncidentService incidentService;
public static void main(String[] args) {
// 去头去尾
for (String string : msgList) {
String substring = string.substring(2, string.length() - 2);
System.out.println("去头去尾字符串："+ substring);
String hexStringWithoutSpaces = substring.replaceAll("\\s+", "");
String asciiString = hexToString(hexStringWithoutSpaces);
System.out.println("16进制解析后的数据："+asciiString);
// //截取前17位
String substring1 = asciiString.substring(0, 17);
System.out.println("VIN："+substring1);
String substring2 = asciiString.substring(17, 30);
System.out.println("时间戳："+substring2);
String substring3 = asciiString.substring(30, 40);
System.out.println("经度：" +substring3);
String substring4 = asciiString.substring(41, 50);
System.out.println("纬度："+ substring4);
String substring5 = asciiString.substring(51, 56);
System.out.println("车速："+ substring5);
String substring6 = asciiString.substring(57, 67);
System.out.println("总里程："+ substring6);
String substring7 = asciiString.substring(68, 73);
System.out.println("总电压："+ substring7);
String pattern = "(.{17})(.{10})(.{9})(.{8})(.{2})";
Pattern compile = Pattern.compile(pattern);
Matcher matcher = compile.matcher(asciiString);
if (matcher.find()) {
for (int i = 1; i < matcher.groupCount(); i++) {
System.out.println("Group "+ i + ":" + matcher.group(i));
}
}
}
CoupletMsgData coupletMsgData = new CoupletMsgData();
List<CoupletMsgData> objects = msgList.stream().map(CoupletMsgData::new).toList();
for (CoupletMsgData object : objects) {
//测试所有的事件
for (String s : list) {
IncidentService bean = SpringUtils.getBean(s);
incidentService.incident(object);
}
}
}
public static String hexToString(String hexString) {
StringBuilder asciiString = new StringBuilder();
for (int i = 0; i < hexString.length(); i += 2) {
String hex = hexString.substring(i, i + 2);
int decimal = Integer.parseInt(hex, 16);
asciiString.append((char) decimal);
}
return asciiString.toString();
}
}

2
couplet-modules/couplet-analyze/pom.xml
View File

@ -12,9 +12,7 @@
<artifactId>couplet-analyze</artifactId>
<packaging>pom</packaging>
<modules>
<module>couplet-analyze-incident</module>
<module>couplet-analyze-msg</module>
<module>couplet-analyze-incident/couplet-incident-server</module>
</modules>
<properties>

6
couplet-modules/couplet-electronic-fence/couplet-electronic-fence-server/src/main/java/com/couplet/map/server/controller/FenceController.java
View File

@ -52,9 +52,9 @@ public class FenceController extends BaseController {
@RequiresPermissions("couplet:fence:fenceAdd")
@Log(title = "电子围栏新增",businessType = BusinessType.INSERT)
public Result<?> fenceInsert(HttpServletRequest request, @RequestBody FenceRequest fenceRequest){
if (!fenceService.checkFenceKeyUnique(fenceRequest.getFenceName())) {
return error("新增参数'" + fenceRequest.getFenceName() + "'失败，参数键名已存在");
}
// if (!fenceService.checkFenceKeyUnique(fenceRequest.getFenceName())) {
// return error("新增参数'" + fenceRequest.getFenceName() + "'失败，参数键名已存在");
// }
fenceService.fenceInsert(request,fenceRequest);
return Result.success("新增成功");
}

6
couplet-modules/couplet-electronic-fence/couplet-electronic-fence-server/src/main/java/com/couplet/map/server/service/impl/FenAndLogoServiceImpl.java
View File

@ -28,7 +28,11 @@ public class FenAndLogoServiceImpl extends ServiceImpl<FenAndLogoMapper, Fence>
@Autowired
private StringRedisTemplate redisTemplate;
/**
* 业务实现:添加中间表
* @param fenceId
* @param logoIds
*/
@Override
public void addBach(Integer fenceId, String[] logoIds) {
fenAndLogoMapper.addBach(fenceId,logoIds);

8
couplet-modules/couplet-electronic-fence/couplet-electronic-fence-server/src/main/java/com/couplet/map/server/service/impl/FenceServiceImpl.java
View File

@ -52,14 +52,22 @@ public class FenceServiceImpl extends ServiceImpl<FenceMapper, Fence> implements
}
/**
* 业务实现:添加围栏
* @param request
* @param fenceRequest
*/
@Override
public void fenceInsert(HttpServletRequest request,FenceRequest fenceRequest) {
//先添加围栏
int a= fenceMapper.insertFence(fenceRequest);
String[] logoIds = fenceRequest.getLogoIds();
String[] parts = new String[0];
for (String logoId : logoIds) {
//把前台传入的字符串分割成数组
parts = logoId.split(",");
//再添加围栏和标识中间表
fenAndLogoService.addBach(fenceRequest.getFenceId(),parts);
}

0
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/service/MqttListen.java
View File