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

Compare commits

base: five-groups:fe7a832714d231aa9d56c23316f4fff79b46845b
Branches Tags
five-groups:server_five_fufanrui
five-groups:server_five_dongxiaodong
five-groups:server_five
five-groups:server_five_liuyunhu
...
compare: five-groups:94a0e48edcd14935737c795b1aab0a0299bdbdf5
Branches Tags
five-groups:server_five_fufanrui
five-groups:server_five_dongxiaodong
five-groups:server_five
five-groups:server_five_liuyunhu

11 Commits
fe7a832714 ... 94a0e48edc

Author SHA1 Message Date
ffr 94a0e48edc ffffff 2024-04-06 16:34:44 +08:00
lijiayao 7aa223c8e4 Merge branch 'server_five_liuyunhu' of https://gitea.qinmian.online/five-groups/five-groups-couplet into server_five 2024-04-06 16:30:45 +08:00
liuyunhu 78f59d7392 优化 2024-04-06 16:30:00 +08:00
dongxiaodong 6798b04fc4 kafka代码 2024-04-06 16:29:50 +08:00
liuyunhu b655ca9c30 打印SQL执行日志 2024-04-06 14:53:29 +08:00
liuyunhu 73342572cc 将远程调用的内容加 自动配置 2024-04-06 14:51:05 +08:00
liuyunhu 812b8d47f8 Kafka demo 2024-04-06 14:25:44 +08:00
lijiayao 957d3976b9 fix: 代码丢失找回 2024-04-06 14:09:42 +08:00
lijiayao a042d30eec Merge branch 'server_five_liuyunhu' of https://gitea.qinmian.online/five-groups/five-groups-couplet into server_five 2024-04-06 14:09:35 +08:00
liuyunhu 858acc75ee Merge branch 'server_five' of https://gitea.qinmian.online/five-groups/five-groups-couplet into server_five_liuyunhu 
# Conflicts:
#	couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java
#	couplet-modules/couplet-business/src/main/resources/mapper/business/SysTroubleMapper.xml
#	couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/service/MqConsumer.java
 2024-04-06 11:38:24 +08:00
liuyunhu e885895016 VS 2024-04-06 11:37:22 +08:00
23 changed files with 958 additions and 569 deletions
Show Stats Expand all files Collapse all files

1
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/RemoteVehicleService.java
View File

@ -54,4 +54,5 @@ public interface RemoteVehicleService {
@GetMapping("onOrOutLineByVIN") @GetMapping("onOrOutLineByVIN")
public Integer onOrOutLineByVIN(@RequestParam("params") String params); public Integer onOrOutLineByVIN(@RequestParam("params") String params);
} }

2
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/factory/RemoteVehicleFallbackFactory.java
View File

@ -54,7 +54,7 @@ public class RemoteVehicleFallbackFactory implements FallbackFactory<RemoteVehic
@Override @Override
public Integer onOrOutLineByVIN(String params) { public Integer onOrOutLineByVIN(String params) {
log.error("车辆服务调用失败:" + cause.getMessage()); log.error("车辆服务调用失败:" + cause.getMessage());
return null; return 0;
} }
}; };
} }

2
couplet-common/couplet-common-business/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 100644
View File

@ -0,0 +1,2 @@
com.couplet.remote.factory.RemoteVehicleFallbackFactory
com.couplet.remote.factory.RemoteVehicleFallbackFactory

22
couplet-common/couplet-common-system/src/main/java/com/couplet/common/system/remote/RemoteCodeService.java
View File

@ -1,22 +0,0 @@
package com.couplet.common.system.remote;
import com.couplet.common.core.constant.ServiceNameConstants;
import com.couplet.common.core.domain.Result;
import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.common.system.remote.factory.RemoteCodeFallbackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
/**
* @author DongXiaoDong
* @version 1.0
* @date 2024/4/4 16:00
* @description
*/
@FeignClient(contextId = "remoteCodeService",value = ServiceNameConstants.BUSINESS_SERVICE, fallbackFactory = RemoteCodeFallbackFactory.class)
public interface RemoteCodeService {
@PostMapping("trouble/insertCode")
public Result<Integer> insertCode(@RequestBody CoupletTroubleCode coupletTroubleCode);
}

30
couplet-common/couplet-common-system/src/main/java/com/couplet/common/system/remote/factory/RemoteCodeFallbackFactory.java
View File

@ -1,30 +0,0 @@
package com.couplet.common.system.remote.factory;
import com.couplet.common.core.domain.Result;
import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.common.system.remote.RemoteCodeService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
/**
* @author DongXiaoDong
* @version 1.0
* @date 2024/4/4 16:03
* @description
*/
@Slf4j
@Component
public class RemoteCodeFallbackFactory implements FallbackFactory<RemoteCodeService> {
@Override
public RemoteCodeService create(Throwable cause) {
log.error("调用日志服务异常：{}", cause.getMessage());
return new RemoteCodeService()
{
@Override
public Result<Integer> insertCode(CoupletTroubleCode coupletTroubleCode) {
return null;
}
};
}
}

1
couplet-common/couplet-common-system/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@ -3,4 +3,3 @@ com.couplet.common.system.remote.factory.RemoteLogFallbackFactory
com.couplet.common.system.remote.factory.RemoteFileFallbackFactory com.couplet.common.system.remote.factory.RemoteFileFallbackFactory
com.couplet.common.system.remote.factory.RemoteDeptFallbackFactory com.couplet.common.system.remote.factory.RemoteDeptFallbackFactory
com.couplet.common.system.remote.factory.RemoteEmployeeFallbackFactory com.couplet.common.system.remote.factory.RemoteEmployeeFallbackFactory
com.couplet.common.system.remote.factory.RemoteCodeFallbackFactory

16
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/consumer/KafkaConsumer.java 100644
View File

@ -0,0 +1,16 @@
package com.couplet.analyze.msg.consumer;
import org.springframework.kafka.annotation.KafkaListener;
/**
* @author DongXiaoDong
* @version 1.0
* @date 2024/4/6 15:37
* @description
*/
public class KafkaConsumer {
@KafkaListener(topics = "test", groupId = "group", properties = {"bootstrap.servers = 39.103.133.136:9092"})
public void getMessage(String msg) {
System.out.println("接收到消息：" + msg);
}
}

728
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelMessage.java
View File

@ -3,7 +3,6 @@ package com.couplet.analyze.msg.model;
import com.couplet.analyze.msg.domain.CoupletMsgData; import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService; import com.couplet.analyze.msg.service.IncidentService;
import com.couplet.common.core.utils.SpringUtils; import com.couplet.common.core.utils.SpringUtils;
import com.couplet.common.system.remote.RemoteCodeService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*; import org.eclipse.paho.client.mqttv3.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate;
@ -29,372 +28,369 @@ import static com.couplet.analyze.msg.contents.MsgContent.CLIENT_ID;
@Component @Component
public class ModelMessage { public class ModelMessage {
@Autowired // @Autowired
private RabbitTemplate rabbitTemplate; // private RabbitTemplate rabbitTemplate;
static ArrayList<String> strings = new ArrayList<>() { // static ArrayList<String> strings = new ArrayList<>() {
{ // {
add("breakdown"); // add("breakdown");
add("electronic-fence"); // add("electronic-fence");
add("real-time-data"); // add("real-time-data");
add("stored-event"); // add("stored-event");
} // }
}; // };
@Autowired
private RemoteCodeService remoteCodeService;
// @Value("${mq.queueName}")
// public String queueName;
// //
// //交换机 //// @Value("${mq.queueName}")
// @Value("${mq.exchangeName}") //// public String queueName;
// public String exchangeName; ////
//// //交换机
//// @Value("${mq.exchangeName}")
//// public String exchangeName;
////
//// //路由键
//// @Value("${mq.routingKey}")
//// public String routingKey;
// //
// //路由键 // @Scheduled(cron = "0/5 * * * * ?")
// @Value("${mq.routingKey}") // public void startMsg() {
// public String routingKey; // try {
// MqttClient mqttClient = new MqttClient(BROKER_URL, CLIENT_ID);
@Scheduled(cron = "0/5 * * * * ?") //
public void startMsg() { // MqttConnectOptions options = new MqttConnectOptions();
try { //
MqttClient mqttClient = new MqttClient(BROKER_URL, CLIENT_ID); // options.setCleanSession(true);
// mqttClient.connect(options);
MqttConnectOptions options = new MqttConnectOptions(); //
// mqttClient.setCallback(new MqttCallback() {
options.setCleanSession(true); // @Override
mqttClient.connect(options); // public void connectionLost(Throwable throwable) {
// log.error("Mqtt[{}-{}]连接断开:[{}]", CLIENT_ID, BROKER_URL, throwable.getMessage(), throwable);
mqttClient.setCallback(new MqttCallback() { // }
@Override //
public void connectionLost(Throwable throwable) { // @Override
log.error("Mqtt[{}-{}]连接断开:[{}]", CLIENT_ID, BROKER_URL, throwable.getMessage(), throwable); // public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
} // //打印接收到的消息和主题
// log.info("主题='{}':消息内容={}", topic, new String(mqttMessage.getPayload()));
@Override //// String str = hexToString(new String(mqttMessage.getPayload()));
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { //// List<CoupletMsgData> coupletMsgDataList = sendMsg(str);
//打印接收到的消息和主题 //
log.info("主题='{}':消息内容={}", topic, new String(mqttMessage.getPayload())); // for (CoupletMsgData msgData : coupletMsgDataList) {
String str = hexToString(new String(mqttMessage.getPayload())); // log.info("解析到车辆数据：{}", msgData);
List<CoupletMsgData> coupletMsgDataList = sendMsg(str); // for (String string : strings) {
// IncidentService incidentService = SpringUtils.getBean(string);
for (CoupletMsgData msgData : coupletMsgDataList) { // incidentService.incident(msgData);
log.info("解析到车辆数据：{}", msgData);
for (String string : strings) {
IncidentService incidentService = SpringUtils.getBean(string);
incidentService.incident(msgData);
}
// 不睡眠，他们也会主动拉取数据，不会影响其他服务
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// throw new RuntimeException(e);
// } // }
} //// 不睡眠，他们也会主动拉取数据，不会影响其他服务
//// try {
} //// Thread.sleep(1000);
//// } catch (InterruptedException e) {
@Override //// throw new RuntimeException(e);
public void deliveryComplete(IMqttDeliveryToken token) { //// }
log.info("消息已投递成功：{}",token); // }
} //
}); // }
mqttClient.subscribe("test",0); //
// @Override
Thread.sleep(1000*60*10); // public void deliveryComplete(IMqttDeliveryToken token) {
} catch (Exception e) { // log.info("消息已投递成功：{}",token);
throw new RuntimeException(e); // }
} // });
} // mqttClient.subscribe("test",0);
//
/** // Thread.sleep(1000*60*10);
* 将16进制字符串转换为ASCII字符串 // } catch (Exception e) {
* @param s 16进制字符串 // throw new RuntimeException(e);
* @return ASCII字符串 // }
*/ // }
public static String hexToString(String s) { //
if (s == null || s.equals("")) { // /**
return null; // * 将16进制字符串转换为ASCII字符串
} // * @param s 16进制字符串
s = s.replace(" ", ""); // * @return ASCII字符串
byte[] baKeyword = new byte[s.length() / 2]; // */
for (int i = 0; i < baKeyword.length; i++) { //// public static String hexToString(String s) {
try { //// if (s == null || s.equals("")) {
baKeyword[i] = (byte) (0xff & Integer.parseInt(s.substring(i * 2, i * 2 + 2), 16)); //// return null;
} catch (Exception e) { //// }
e.printStackTrace(); //// s = s.replace(" ", "");
} //// byte[] baKeyword = new byte[s.length() / 2];
} //// for (int i = 0; i < baKeyword.length; i++) {
try { //// try {
s = new String(baKeyword, StandardCharsets.UTF_8); //// baKeyword[i] = (byte) (0xff & Integer.parseInt(s.substring(i * 2, i * 2 + 2), 16));
} catch (Exception e1) { //// } catch (Exception e) {
e1.printStackTrace(); //// e.printStackTrace();
return s; //// }
} //// }
return s; //// try {
} //// s = new String(baKeyword, StandardCharsets.UTF_8);
//// } catch (Exception e1) {
public static List<CoupletMsgData> sendMsg(String str) { //// e1.printStackTrace();
List<CoupletMsgData> coupletMsgDataList = new ArrayList<>(); //// return s;
CoupletMsgData coupletMsgData = new CoupletMsgData(); //// }
//// return s;
coupletMsgData.setVin(str.substring(1,18)); //// }
////
log.info("vin=="+coupletMsgData.getVin()); //// public static List<CoupletMsgData> sendMsg(String str) {
//// List<CoupletMsgData> coupletMsgDataList = new ArrayList<>();
//时间 //// CoupletMsgData coupletMsgData = new CoupletMsgData();
String tim =str.substring(18,31); ////
long timestamp = Long.parseLong(tim); //// coupletMsgData.setVin(str.substring(1,18));
////
Date date = new Date(timestamp); //// log.info("vin=="+coupletMsgData.getVin());
coupletMsgData.setCreateTime(date); ////
//// //时间
//经度 //// String tim =str.substring(18,31);
String lt = str.substring(31,42); //// long timestamp = Long.parseLong(tim);
// 如果末尾是零，则舍去 ////
int endIndex = lt.length() - 1; //// Date date = new Date(timestamp);
while (lt.charAt(endIndex) == '0'){ //// coupletMsgData.setCreateTime(date);
endIndex--; ////
} //// //经度
//// String lt = str.substring(31,42);
String longitude = lt.substring(0, endIndex + 1); //// // 如果末尾是零，则舍去
coupletMsgData.setLongitude(longitude); //// int endIndex = lt.length() - 1;
//// while (lt.charAt(endIndex) == '0'){
//维度 //// endIndex--;
String latitudeIndex =str.substring(42,52); //// }
int endIndexT = latitudeIndex.length() - 1; ////
while (latitudeIndex.charAt(endIndexT) == '0'){ //// String longitude = lt.substring(0, endIndex + 1);
endIndexT--; //// coupletMsgData.setLongitude(longitude);
} ////
//// //维度
String latitude = latitudeIndex.substring(0, endIndexT + 1); //// String latitudeIndex =str.substring(42,52);
coupletMsgData.setLatitude(latitude); //// int endIndexT = latitudeIndex.length() - 1;
//// while (latitudeIndex.charAt(endIndexT) == '0'){
//速度speed //// endIndexT--;
String speed =str.substring(52,58); //// }
coupletMsgData.setSpeed(speed); ////
//// String latitude = latitudeIndex.substring(0, endIndexT + 1);
//里程 //// coupletMsgData.setLatitude(latitude);
BigDecimal mileage= new BigDecimal(str.substring(58,69)); ////
mileage=mileage.stripTrailingZeros(); //// //速度speed
coupletMsgData.setMileage(mileage); //// String speed =str.substring(52,58);
//// coupletMsgData.setSpeed(speed);
//总电压 ////
String voltage =str.substring(69,75); //// //里程
while (voltage.endsWith("0")) { // BigDecimal mileage= new BigDecimal(str.substring(58,69));
voltage = voltage.substring(0, voltage.length() - 1); // 去除末尾的零 // mileage=mileage.stripTrailingZeros();
} // coupletMsgData.setMileage(mileage);
coupletMsgData.setVoltage(voltage); //
// //总电压
//总电流 // String voltage =str.substring(69,75);
String current =str.substring(75,80); // while (voltage.endsWith("0")) {
while (current.endsWith("0")){ // voltage = voltage.substring(0, voltage.length() - 1); // 去除末尾的零
current=current.substring(0,current.length()-1); // }
} // coupletMsgData.setVoltage(voltage);
coupletMsgData.setCurrent(current); //
// //总电流
//绝缘电阻 resistance // String current =str.substring(75,80);
String res =str.substring(80,89); // while (current.endsWith("0")){
String resistance = res.substring(0, 5); // current=current.substring(0,current.length()-1);
coupletMsgData.setResistance(resistance); // }
// coupletMsgData.setCurrent(current);
//档位 //
String gear =str.substring(89,90); // //绝缘电阻 resistance
coupletMsgData.setGear(gear); // String res =str.substring(80,89);
// String resistance = res.substring(0, 5);
//accelerationPedal 加速踏板行程值 // coupletMsgData.setResistance(resistance);
String accelerationPedal =str.substring(90,91); //
coupletMsgData.setAccelerationPedal(accelerationPedal); // //档位
// String gear =str.substring(89,90);
//brakePedal 制动踏板行程值 // coupletMsgData.setGear(gear);
String brakePedal =str.substring(92,93); //
coupletMsgData.setBrakePedal(brakePedal); // //accelerationPedal 加速踏板行程值
// String accelerationPedal =str.substring(90,91);
//fuelConsumptionRate 燃料消耗率 // coupletMsgData.setAccelerationPedal(accelerationPedal);
String fuelConsumptionRate =str.substring(94,99); //
coupletMsgData.setFuelConsumptionRate(fuelConsumptionRate); // //brakePedal 制动踏板行程值
// String brakePedal =str.substring(92,93);
//motorControllerTemperature 电机控制器温度 // coupletMsgData.setBrakePedal(brakePedal);
String motorControllerTemperature =str.substring(99,105); //
while (motorControllerTemperature.endsWith("0")){ // //fuelConsumptionRate 燃料消耗率
motorControllerTemperature=motorControllerTemperature.substring(0,motorControllerTemperature.length()-1); // String fuelConsumptionRate =str.substring(94,99);
} // coupletMsgData.setFuelConsumptionRate(fuelConsumptionRate);
coupletMsgData.setMotorControllerTemperature(motorControllerTemperature); //
// //motorControllerTemperature 电机控制器温度
//motorSpeed 电机转速 // String motorControllerTemperature =str.substring(99,105);
String motorSpeed =str.substring(105,110); // while (motorControllerTemperature.endsWith("0")){
coupletMsgData.setMotorSpeed(motorSpeed); // motorControllerTemperature=motorControllerTemperature.substring(0,motorControllerTemperature.length()-1);
// }
//motorTorque 电机转矩 // coupletMsgData.setMotorControllerTemperature(motorControllerTemperature);
String motorTorque =str.substring(110,114); //
while (motorTorque.endsWith("0")){ // //motorSpeed 电机转速
motorTorque=motorTorque.substring(0,motorTorque.length()-1); // String motorSpeed =str.substring(105,110);
} // coupletMsgData.setMotorSpeed(motorSpeed);
coupletMsgData.setMotorTorque(motorTorque); //
// //motorTorque 电机转矩
//motorTemperature 电机温度 // String motorTorque =str.substring(110,114);
String motorTemperature =str.substring(114,120); // while (motorTorque.endsWith("0")){
while (motorTemperature.endsWith("0")){ // motorTorque=motorTorque.substring(0,motorTorque.length()-1);
motorTemperature=motorTemperature.substring(0,motorTemperature.length()-1); // }
} // coupletMsgData.setMotorTorque(motorTorque);
coupletMsgData.setMotorTemperature(motorTemperature); //
// //motorTemperature 电机温度
//motorVoltage 电机电压 // String motorTemperature =str.substring(114,120);
String motorVoltage =str.substring(120,125); // while (motorTemperature.endsWith("0")){
while (motorVoltage.endsWith("0")){ // motorTemperature=motorTemperature.substring(0,motorTemperature.length()-1);
motorVoltage=motorVoltage.substring(0,motorVoltage.length()-1); // }
} // coupletMsgData.setMotorTemperature(motorTemperature);
coupletMsgData.setMotorVoltage(motorVoltage); //
// //motorVoltage 电机电压
//motorCurrent 电机电流 // String motorVoltage =str.substring(120,125);
String motorCurrent =str.substring(125,133); // while (motorVoltage.endsWith("0")){
while (motorCurrent.endsWith("0")){ // motorVoltage=motorVoltage.substring(0,motorVoltage.length()-1);
motorCurrent=motorCurrent.substring(0,motorCurrent.length()-1); // }
} // coupletMsgData.setMotorVoltage(motorVoltage);
coupletMsgData.setMotorCurrent(motorCurrent); //
// //motorCurrent 电机电流
//remainingBattery 动力电池剩余电量SOC // String motorCurrent =str.substring(125,133);
BigDecimal remainingBattery = new BigDecimal(str.substring(133,138)); // while (motorCurrent.endsWith("0")){
coupletMsgData.setRemainingBattery(remainingBattery); // motorCurrent=motorCurrent.substring(0,motorCurrent.length()-1);
// }
//maximumFeedbackPower 当前状态允许的最大反馈功率 // coupletMsgData.setMotorCurrent(motorCurrent);
String maximumFeedbackPower =str.substring(139,144); //
while (maximumFeedbackPower.endsWith("0")){ // //remainingBattery 动力电池剩余电量SOC
maximumFeedbackPower=maximumFeedbackPower.substring(0,maximumFeedbackPower.length()-1); // BigDecimal remainingBattery = new BigDecimal(str.substring(133,138));
} // coupletMsgData.setRemainingBattery(remainingBattery);
coupletMsgData.setMaximumFeedbackPower(maximumFeedbackPower); //
// //maximumFeedbackPower 当前状态允许的最大反馈功率
//maximumDischargePower 当前状态允许最大放电功率 // String maximumFeedbackPower =str.substring(139,144);
String maximumDischargePower =str.substring(145,151); // while (maximumFeedbackPower.endsWith("0")){
while (maximumDischargePower.endsWith("0")){ // maximumFeedbackPower=maximumFeedbackPower.substring(0,maximumFeedbackPower.length()-1);
maximumDischargePower=maximumDischargePower.substring(0,maximumDischargePower.length()-1); // }
} // coupletMsgData.setMaximumFeedbackPower(maximumFeedbackPower);
coupletMsgData.setMaximumDischargePower(maximumDischargePower); //
// //maximumDischargePower 当前状态允许最大放电功率
//selfCheckCounter BMS自检计数器 // String maximumDischargePower =str.substring(145,151);
String selfCheckCounter =str.substring(151,153); // while (maximumDischargePower.endsWith("0")){
String selfCheckCounterReplace = selfCheckCounter.replace("0", ""); // maximumDischargePower=maximumDischargePower.substring(0,maximumDischargePower.length()-1);
coupletMsgData.setSelfCheckCounter(selfCheckCounterReplace); // }
// coupletMsgData.setMaximumDischargePower(maximumDischargePower);
//totalBatteryCurrent 动力电池充放电电流 //
String totalBatteryCurrent =str.substring(153,158); // //selfCheckCounter BMS自检计数器
while (totalBatteryCurrent.endsWith("0")){ // String selfCheckCounter =str.substring(151,153);
totalBatteryCurrent=totalBatteryCurrent.substring(0,totalBatteryCurrent.length()-1); // String selfCheckCounterReplace = selfCheckCounter.replace("0", "");
} // coupletMsgData.setSelfCheckCounter(selfCheckCounterReplace);
coupletMsgData.setTotalBatteryCurrent(totalBatteryCurrent); //
// //totalBatteryCurrent 动力电池充放电电流
//totalBatteryVoltage 动力电池负载端总电压V3 // String totalBatteryCurrent =str.substring(153,158);
String totalBatteryVoltage =str.substring(158,164); // while (totalBatteryCurrent.endsWith("0")){
while (totalBatteryVoltage.endsWith("0")){ // totalBatteryCurrent=totalBatteryCurrent.substring(0,totalBatteryCurrent.length()-1);
totalBatteryVoltage=totalBatteryVoltage.substring(0,totalBatteryVoltage.length()-1); // }
} // coupletMsgData.setTotalBatteryCurrent(totalBatteryCurrent);
coupletMsgData.setTotalBatteryVoltage(totalBatteryVoltage); //
// //totalBatteryVoltage 动力电池负载端总电压V3
//singleBatteryMaxVoltage 单次最大电压 // String totalBatteryVoltage =str.substring(158,164);
String singleBatteryMaxVoltage =str.substring(164,168); // while (totalBatteryVoltage.endsWith("0")){
while (singleBatteryMaxVoltage.endsWith("0")){ // totalBatteryVoltage=totalBatteryVoltage.substring(0,totalBatteryVoltage.length()-1);
singleBatteryMaxVoltage=singleBatteryMaxVoltage.substring(0,singleBatteryMaxVoltage.length()-1); // }
} // coupletMsgData.setTotalBatteryVoltage(totalBatteryVoltage);
coupletMsgData.setSingleBatteryMaxVoltage(singleBatteryMaxVoltage); //
// //singleBatteryMaxVoltage 单次最大电压
//singleBatteryMinVoltage 单体电池最低电压 // String singleBatteryMaxVoltage =str.substring(164,168);
String singleBatteryMinVoltage =str.substring(168,172); // while (singleBatteryMaxVoltage.endsWith("0")){
while (singleBatteryMinVoltage.endsWith("0")){ // singleBatteryMaxVoltage=singleBatteryMaxVoltage.substring(0,singleBatteryMaxVoltage.length()-1);
singleBatteryMinVoltage=singleBatteryMinVoltage.substring(0,singleBatteryMinVoltage.length()-1); // }
} // coupletMsgData.setSingleBatteryMaxVoltage(singleBatteryMaxVoltage);
//
coupletMsgData.setSingleBatteryMinVoltage(singleBatteryMinVoltage); // //singleBatteryMinVoltage 单体电池最低电压
// String singleBatteryMinVoltage =str.substring(168,172);
//singleBatteryMaxTemperature 单体电池最高温度 // while (singleBatteryMinVoltage.endsWith("0")){
String singleBatteryMaxTemperature =str.substring(172,178); // singleBatteryMinVoltage=singleBatteryMinVoltage.substring(0,singleBatteryMinVoltage.length()-1);
while (singleBatteryMaxTemperature.endsWith("0")){ // }
singleBatteryMaxTemperature=singleBatteryMaxTemperature.substring(0,singleBatteryMaxTemperature.length()-1); //
} // coupletMsgData.setSingleBatteryMinVoltage(singleBatteryMinVoltage);
coupletMsgData.setSingleBatteryMaxTemperature(singleBatteryMaxTemperature); //
// //singleBatteryMaxTemperature 单体电池最高温度
//singleBatteryMinTemperature 单体电池最低温度 // String singleBatteryMaxTemperature =str.substring(172,178);
String singleBatteryMinTemperature =str.substring(178,184); // while (singleBatteryMaxTemperature.endsWith("0")){
while (singleBatteryMinTemperature.endsWith("0")){ // singleBatteryMaxTemperature=singleBatteryMaxTemperature.substring(0,singleBatteryMaxTemperature.length()-1);
singleBatteryMinTemperature=singleBatteryMinTemperature.substring(0,singleBatteryMinTemperature.length()-1); // }
} // coupletMsgData.setSingleBatteryMaxTemperature(singleBatteryMaxTemperature);
coupletMsgData.setSingleBatteryMinTemperature(singleBatteryMinTemperature); //
// //singleBatteryMinTemperature 单体电池最低温度
//availableBatteryCapacity 可用电池容量 // String singleBatteryMinTemperature =str.substring(178,184);
String availableBatteryCapacity =str.substring(184,190); // while (singleBatteryMinTemperature.endsWith("0")){
while (availableBatteryCapacity.endsWith("0")){ // singleBatteryMinTemperature=singleBatteryMinTemperature.substring(0,singleBatteryMinTemperature.length()-1);
availableBatteryCapacity=availableBatteryCapacity.substring(0,availableBatteryCapacity.length()-1); // }
} // coupletMsgData.setSingleBatteryMinTemperature(singleBatteryMinTemperature);
coupletMsgData.setAvailableBatteryCapacity(availableBatteryCapacity); //
// //availableBatteryCapacity 可用电池容量
//vehicleStatus 车辆状态 // String availableBatteryCapacity =str.substring(184,190);
int vehicleStatus = Integer.parseInt(str.substring(190,191)); // while (availableBatteryCapacity.endsWith("0")){
coupletMsgData.setVehicleStatus(vehicleStatus); // availableBatteryCapacity=availableBatteryCapacity.substring(0,availableBatteryCapacity.length()-1);
// }
//chargingStatus 充电状态 // coupletMsgData.setAvailableBatteryCapacity(availableBatteryCapacity);
int chargingStatus = Integer.parseInt(str.substring(191,192)); //
coupletMsgData.setChargingStatus(chargingStatus); // //vehicleStatus 车辆状态
// int vehicleStatus = Integer.parseInt(str.substring(190,191));
//operatingStatus 运行状态 // coupletMsgData.setVehicleStatus(vehicleStatus);
int operatingStatus = Integer.parseInt(str.substring(192,193)); //
coupletMsgData.setOperatingStatus(operatingStatus); // //chargingStatus 充电状态
// int chargingStatus = Integer.parseInt(str.substring(191,192));
//socStatus SOC // coupletMsgData.setChargingStatus(chargingStatus);
int socStatus = Integer.parseInt(str.substring(193,194)); //
coupletMsgData.setSocStatus(socStatus); // //operatingStatus 运行状态
// int operatingStatus = Integer.parseInt(str.substring(192,193));
//chargingEnergyStorageStatus 可充电储能装置工作状态 // coupletMsgData.setOperatingStatus(operatingStatus);
int chargingEnergyStorageStatus = Integer.parseInt(str.substring(194,195)); //
coupletMsgData.setChargingEnergyStorageStatus(chargingEnergyStorageStatus); // //socStatus SOC
// int socStatus = Integer.parseInt(str.substring(193,194));
//driveMotorStatus 驱动电机状态 // coupletMsgData.setSocStatus(socStatus);
int driveMotorStatus = Integer.parseInt(str.substring(195,196)); //
coupletMsgData.setDriveMotorStatus(driveMotorStatus); // //chargingEnergyStorageStatus 可充电储能装置工作状态
// int chargingEnergyStorageStatus = Integer.parseInt(str.substring(194,195));
//positionStatus 定位是否有效 // coupletMsgData.setChargingEnergyStorageStatus(chargingEnergyStorageStatus);
int positionStatus = Integer.parseInt(str.substring(196,197)); //
coupletMsgData.setPositionStatus(positionStatus); // //driveMotorStatus 驱动电机状态
// int driveMotorStatus = Integer.parseInt(str.substring(195,196));
//easStatus EAS(汽车防盗系统)状态 // coupletMsgData.setDriveMotorStatus(driveMotorStatus);
int easStatus = Integer.parseInt(str.substring(197,198)); //
coupletMsgData.setEasStatus(easStatus); // //positionStatus 定位是否有效
// int positionStatus = Integer.parseInt(str.substring(196,197));
//ptcStatus PTC(电动加热器)状态 // coupletMsgData.setPositionStatus(positionStatus);
int ptcStatus = Integer.parseInt(str.substring(198,199)); //
coupletMsgData.setPtcStatus(ptcStatus); // //easStatus EAS(汽车防盗系统)状态
// int easStatus = Integer.parseInt(str.substring(197,198));
//epsStatus // coupletMsgData.setEasStatus(easStatus);
int epsStatus = Integer.parseInt(str.substring(199,200)); //
coupletMsgData.setEpsStatus(epsStatus); // //ptcStatus PTC(电动加热器)状态
// int ptcStatus = Integer.parseInt(str.substring(198,199));
//absStatus EPS(电动助力系统)状态 // coupletMsgData.setPtcStatus(ptcStatus);
int absStatus = Integer.parseInt(str.substring(200,201)); //
coupletMsgData.setAbsStatus(absStatus); // //epsStatus
// int epsStatus = Integer.parseInt(str.substring(199,200));
//mcuStatus MCU(电机/逆变器)状态 // coupletMsgData.setEpsStatus(epsStatus);
int mcuStatus = Integer.parseInt(str.substring(201,202)); //
coupletMsgData.setMcuStatus(mcuStatus); // //absStatus EPS(电动助力系统)状态
// int absStatus = Integer.parseInt(str.substring(200,201));
//heatingStatus 动力电池加热状态 // coupletMsgData.setAbsStatus(absStatus);
int heatingStatus = Integer.parseInt(str.substring(202,203)); //
coupletMsgData.setHeatingStatus(heatingStatus); // //mcuStatus MCU(电机/逆变器)状态
// int mcuStatus = Integer.parseInt(str.substring(201,202));
//batteryStatus 动力电池当前状态 // coupletMsgData.setMcuStatus(mcuStatus);
int batteryStatus = Integer.parseInt(str.substring(203,204)); //
coupletMsgData.setBatteryStatus(batteryStatus); // //heatingStatus 动力电池加热状态
// int heatingStatus = Integer.parseInt(str.substring(202,203));
//batteryInsulationStatus 动力电池保温状态 // coupletMsgData.setHeatingStatus(heatingStatus);
int batteryInsulationStatus = Integer.parseInt(str.substring(204,205)); //
coupletMsgData.setBatteryInsulationStatus(batteryInsulationStatus); // //batteryStatus 动力电池当前状态
// int batteryStatus = Integer.parseInt(str.substring(203,204));
//dcdcStatus DCDC(电力交换系统)状态 // coupletMsgData.setBatteryStatus(batteryStatus);
int dcdcStatus = Integer.parseInt(str.substring(205,206)); //
coupletMsgData.setDcdcStatus(dcdcStatus); // //batteryInsulationStatus 动力电池保温状态
// int batteryInsulationStatus = Integer.parseInt(str.substring(204,205));
//chgStatus CHG(充电机)状态 // coupletMsgData.setBatteryInsulationStatus(batteryInsulationStatus);
int chgStatus = Integer.parseInt(str.substring(206,207)); //
coupletMsgData.setChgStatus(chgStatus); // //dcdcStatus DCDC(电力交换系统)状态
// int dcdcStatus = Integer.parseInt(str.substring(205,206));
coupletMsgDataList.add(coupletMsgData); // coupletMsgData.setDcdcStatus(dcdcStatus);
//
return coupletMsgDataList; // //chgStatus CHG(充电机)状态
} // int chgStatus = Integer.parseInt(str.substring(206,207));
// coupletMsgData.setChgStatus(chgStatus);
//
// coupletMsgDataList.add(coupletMsgData);
//
// return coupletMsgDataList;
// }
} }

86
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java 100644
View File

@ -0,0 +1,86 @@
package com.couplet.analyze.msg.model;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import com.couplet.common.core.utils.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import static com.couplet.analyze.msg.utils.MsgUtils.hexToString;
import static com.couplet.analyze.msg.utils.MsgUtils.sendMsg;
/**
* @author DongXiaoDong
* @version 1.0
* @date 2024/4/6 14:04
* @description
*/
@Component
@Slf4j
public class ModelsKafkaMessage {
private static final String TOPIC_NAME = "online";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
static ArrayList<String> strings = new ArrayList<>() {
{
add("breakdown");
add("electronic-fence");
add("real-time-data");
add("stored-event");
}
};
/**
* 消费者配置
* @return
*/
@Scheduled(fixedDelay = 50)
private static void consumerMessages() {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//创建消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
try {
//订阅主题
consumer.subscribe(Collections.singletonList(TOPIC_NAME));
//持续消费消息
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {
System.out.println("接收到的数据：" + record.value());
String str = hexToString(record.value());
List<CoupletMsgData> coupletMsgDataList = sendMsg(str);
for (CoupletMsgData msgData : coupletMsgDataList) {
log.info("解析到车辆数据：{}", msgData);
for (String string : strings) {
IncidentService incidentService = SpringUtils.getBean(string);
incidentService.incident(msgData);
}
}
});
}
} catch (Exception e) {
e.printStackTrace();
} finally {
consumer.close();
}
}
}

317
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/MsgUtils.java 100644
View File

@ -0,0 +1,317 @@
package com.couplet.analyze.msg.utils;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import lombok.extern.slf4j.Slf4j;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
/**
* @author DongXiaoDong
* @version 1.0
* @date 2024/4/6 14:12
* @description
*/
@Slf4j
public class MsgUtils {
/**
* 将16进制字符串转换为ASCII字符串
* @param s 16进制字符串
* @return ASCII字符串
*/
public static String hexToString(String s) {
if (s == null || s.equals("")) {
return null;
}
s = s.replace(" ", "");
byte[] baKeyword = new byte[s.length() / 2];
for (int i = 0; i < baKeyword.length; i++) {
try {
baKeyword[i] = (byte) (0xff & Integer.parseInt(s.substring(i * 2, i * 2 + 2), 16));
} catch (Exception e) {
e.printStackTrace();
}
}
try {
s = new String(baKeyword, StandardCharsets.UTF_8);
} catch (Exception e1) {
e1.printStackTrace();
return s;
}
return s;
}
/**
* 将解析后的数据进行截取封装
* @param str
* @return
*/
public static List<CoupletMsgData> sendMsg(String str) {
List<CoupletMsgData> coupletMsgDataList = new ArrayList<>();
CoupletMsgData coupletMsgData = new CoupletMsgData();
coupletMsgData.setVin(str.substring(1,18));
log.info("vin=="+coupletMsgData.getVin());
//时间
String tim =str.substring(18,31);
long timestamp = Long.parseLong(tim);
Date date = new Date(timestamp);
coupletMsgData.setCreateTime(date);
//经度
String lt = str.substring(31,42);
// 如果末尾是零，则舍去
int endIndex = lt.length() - 1;
while (lt.charAt(endIndex) == '0'){
endIndex--;
}
String longitude = lt.substring(0, endIndex + 1);
coupletMsgData.setLongitude(longitude);
//维度
String latitudeIndex =str.substring(42,52);
int endIndexT = latitudeIndex.length() - 1;
while (latitudeIndex.charAt(endIndexT) == '0'){
endIndexT--;
}
String latitude = latitudeIndex.substring(0, endIndexT + 1);
coupletMsgData.setLatitude(latitude);
//速度speed
String speed =str.substring(52,58);
coupletMsgData.setSpeed(speed);
//里程
BigDecimal mileage= new BigDecimal(str.substring(58,69));
mileage=mileage.stripTrailingZeros();
coupletMsgData.setMileage(mileage);
//总电压
String voltage =str.substring(69,75);
while (voltage.endsWith("0")) {
voltage = voltage.substring(0, voltage.length() - 1); // 去除末尾的零
}
coupletMsgData.setVoltage(voltage);
//总电流
String current =str.substring(75,80);
while (current.endsWith("0")){
current=current.substring(0,current.length()-1);
}
coupletMsgData.setCurrent(current);
//绝缘电阻 resistance
String res =str.substring(80,89);
String resistance = res.substring(0, 5);
coupletMsgData.setResistance(resistance);
//档位
String gear =str.substring(89,90);
coupletMsgData.setGear(gear);
//accelerationPedal 加速踏板行程值
String accelerationPedal =str.substring(90,91);
coupletMsgData.setAccelerationPedal(accelerationPedal);
//brakePedal 制动踏板行程值
String brakePedal =str.substring(92,93);
coupletMsgData.setBrakePedal(brakePedal);
//fuelConsumptionRate 燃料消耗率
String fuelConsumptionRate =str.substring(94,99);
coupletMsgData.setFuelConsumptionRate(fuelConsumptionRate);
//motorControllerTemperature 电机控制器温度
String motorControllerTemperature =str.substring(99,105);
while (motorControllerTemperature.endsWith("0")){
motorControllerTemperature=motorControllerTemperature.substring(0,motorControllerTemperature.length()-1);
}
coupletMsgData.setMotorControllerTemperature(motorControllerTemperature);
//motorSpeed 电机转速
String motorSpeed =str.substring(105,110);
coupletMsgData.setMotorSpeed(motorSpeed);
//motorTorque 电机转矩
String motorTorque =str.substring(110,114);
while (motorTorque.endsWith("0")){
motorTorque=motorTorque.substring(0,motorTorque.length()-1);
}
coupletMsgData.setMotorTorque(motorTorque);
//motorTemperature 电机温度
String motorTemperature =str.substring(114,120);
while (motorTemperature.endsWith("0")){
motorTemperature=motorTemperature.substring(0,motorTemperature.length()-1);
}
coupletMsgData.setMotorTemperature(motorTemperature);
//motorVoltage 电机电压
String motorVoltage =str.substring(120,125);
while (motorVoltage.endsWith("0")){
motorVoltage=motorVoltage.substring(0,motorVoltage.length()-1);
}
coupletMsgData.setMotorVoltage(motorVoltage);
//motorCurrent 电机电流
String motorCurrent =str.substring(125,133);
while (motorCurrent.endsWith("0")){
motorCurrent=motorCurrent.substring(0,motorCurrent.length()-1);
}
coupletMsgData.setMotorCurrent(motorCurrent);
//remainingBattery 动力电池剩余电量SOC
BigDecimal remainingBattery = new BigDecimal(str.substring(133,138));
coupletMsgData.setRemainingBattery(remainingBattery);
//maximumFeedbackPower 当前状态允许的最大反馈功率
String maximumFeedbackPower =str.substring(139,144);
while (maximumFeedbackPower.endsWith("0")){
maximumFeedbackPower=maximumFeedbackPower.substring(0,maximumFeedbackPower.length()-1);
}
coupletMsgData.setMaximumFeedbackPower(maximumFeedbackPower);
//maximumDischargePower 当前状态允许最大放电功率
String maximumDischargePower =str.substring(145,151);
while (maximumDischargePower.endsWith("0")){
maximumDischargePower=maximumDischargePower.substring(0,maximumDischargePower.length()-1);
}
coupletMsgData.setMaximumDischargePower(maximumDischargePower);
//selfCheckCounter BMS自检计数器
String selfCheckCounter =str.substring(151,153);
String selfCheckCounterReplace = selfCheckCounter.replace("0", "");
coupletMsgData.setSelfCheckCounter(selfCheckCounterReplace);
//totalBatteryCurrent 动力电池充放电电流
String totalBatteryCurrent =str.substring(153,158);
while (totalBatteryCurrent.endsWith("0")){
totalBatteryCurrent=totalBatteryCurrent.substring(0,totalBatteryCurrent.length()-1);
}
coupletMsgData.setTotalBatteryCurrent(totalBatteryCurrent);
//totalBatteryVoltage 动力电池负载端总电压V3
String totalBatteryVoltage =str.substring(158,164);
while (totalBatteryVoltage.endsWith("0")){
totalBatteryVoltage=totalBatteryVoltage.substring(0,totalBatteryVoltage.length()-1);
}
coupletMsgData.setTotalBatteryVoltage(totalBatteryVoltage);
//singleBatteryMaxVoltage 单次最大电压
String singleBatteryMaxVoltage =str.substring(164,168);
while (singleBatteryMaxVoltage.endsWith("0")){
singleBatteryMaxVoltage=singleBatteryMaxVoltage.substring(0,singleBatteryMaxVoltage.length()-1);
}
coupletMsgData.setSingleBatteryMaxVoltage(singleBatteryMaxVoltage);
//singleBatteryMinVoltage 单体电池最低电压
String singleBatteryMinVoltage =str.substring(168,172);
while (singleBatteryMinVoltage.endsWith("0")){
singleBatteryMinVoltage=singleBatteryMinVoltage.substring(0,singleBatteryMinVoltage.length()-1);
}
coupletMsgData.setSingleBatteryMinVoltage(singleBatteryMinVoltage);
//singleBatteryMaxTemperature 单体电池最高温度
String singleBatteryMaxTemperature =str.substring(172,178);
while (singleBatteryMaxTemperature.endsWith("0")){
singleBatteryMaxTemperature=singleBatteryMaxTemperature.substring(0,singleBatteryMaxTemperature.length()-1);
}
coupletMsgData.setSingleBatteryMaxTemperature(singleBatteryMaxTemperature);
//singleBatteryMinTemperature 单体电池最低温度
String singleBatteryMinTemperature =str.substring(178,184);
while (singleBatteryMinTemperature.endsWith("0")){
singleBatteryMinTemperature=singleBatteryMinTemperature.substring(0,singleBatteryMinTemperature.length()-1);
}
coupletMsgData.setSingleBatteryMinTemperature(singleBatteryMinTemperature);
//availableBatteryCapacity 可用电池容量
String availableBatteryCapacity =str.substring(184,190);
while (availableBatteryCapacity.endsWith("0")){
availableBatteryCapacity=availableBatteryCapacity.substring(0,availableBatteryCapacity.length()-1);
}
coupletMsgData.setAvailableBatteryCapacity(availableBatteryCapacity);
//vehicleStatus 车辆状态
int vehicleStatus = Integer.parseInt(str.substring(190,191));
coupletMsgData.setVehicleStatus(vehicleStatus);
//chargingStatus 充电状态
int chargingStatus = Integer.parseInt(str.substring(191,192));
coupletMsgData.setChargingStatus(chargingStatus);
//operatingStatus 运行状态
int operatingStatus = Integer.parseInt(str.substring(192,193));
coupletMsgData.setOperatingStatus(operatingStatus);
//socStatus SOC
int socStatus = Integer.parseInt(str.substring(193,194));
coupletMsgData.setSocStatus(socStatus);
//chargingEnergyStorageStatus 可充电储能装置工作状态
int chargingEnergyStorageStatus = Integer.parseInt(str.substring(194,195));
coupletMsgData.setChargingEnergyStorageStatus(chargingEnergyStorageStatus);
//driveMotorStatus 驱动电机状态
int driveMotorStatus = Integer.parseInt(str.substring(195,196));
coupletMsgData.setDriveMotorStatus(driveMotorStatus);
//positionStatus 定位是否有效
int positionStatus = Integer.parseInt(str.substring(196,197));
coupletMsgData.setPositionStatus(positionStatus);
//easStatus EAS(汽车防盗系统)状态
int easStatus = Integer.parseInt(str.substring(197,198));
coupletMsgData.setEasStatus(easStatus);
//ptcStatus PTC(电动加热器)状态
int ptcStatus = Integer.parseInt(str.substring(198,199));
coupletMsgData.setPtcStatus(ptcStatus);
//epsStatus
int epsStatus = Integer.parseInt(str.substring(199,200));
coupletMsgData.setEpsStatus(epsStatus);
//absStatus EPS(电动助力系统)状态
int absStatus = Integer.parseInt(str.substring(200,201));
coupletMsgData.setAbsStatus(absStatus);
//mcuStatus MCU(电机/逆变器)状态
int mcuStatus = Integer.parseInt(str.substring(201,202));
coupletMsgData.setMcuStatus(mcuStatus);
//heatingStatus 动力电池加热状态
int heatingStatus = Integer.parseInt(str.substring(202,203));
coupletMsgData.setHeatingStatus(heatingStatus);
//batteryStatus 动力电池当前状态
int batteryStatus = Integer.parseInt(str.substring(203,204));
coupletMsgData.setBatteryStatus(batteryStatus);
//batteryInsulationStatus 动力电池保温状态
int batteryInsulationStatus = Integer.parseInt(str.substring(204,205));
coupletMsgData.setBatteryInsulationStatus(batteryInsulationStatus);
//dcdcStatus DCDC(电力交换系统)状态
int dcdcStatus = Integer.parseInt(str.substring(205,206));
coupletMsgData.setDcdcStatus(dcdcStatus);
//chgStatus CHG(充电机)状态
int chgStatus = Integer.parseInt(str.substring(206,207));
coupletMsgData.setChgStatus(chgStatus);
coupletMsgDataList.add(coupletMsgData);
return coupletMsgDataList;
}
}

16
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java
View File

@ -75,22 +75,6 @@ public class SysTroubleController extends BaseController {
return success(); return success();
} }
/**
* 故障数据入库
* @param coupletTroubleCode
* @return
*/
@PostMapping("insertCode")
public Result<Integer> insertCode(@RequestBody CoupletTroubleCode coupletTroubleCode){
long start = System.currentTimeMillis();
int i = troubleService.insertMsgResq(coupletTroubleCode);
long end = System.currentTimeMillis();
log.info("记录异常信息成功,耗时：{}",(end-start));
return Result.success(i);
}
/** /**
* 清空故障码信息 * 清空故障码信息
*/ */

1
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java
View File

@ -3,6 +3,7 @@ package com.couplet.business.server.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.couplet.common.domain.CoupletTroubleCode; import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.common.domain.CoupletTroubleGrade; import com.couplet.common.domain.CoupletTroubleGrade;
import com.couplet.common.domain.request.TroubleResp; import com.couplet.common.domain.request.TroubleResp;
import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Mapper;

2
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java
View File

@ -25,7 +25,5 @@ public interface SysTroubleService extends IService<CoupletTroubleCode> {
*/ */
void newFaultData(CoupletTroubleCode code); void newFaultData(CoupletTroubleCode code);
int insertMsgResq(CoupletTroubleCode coupletTroubleCode);
void cleanTroubleCode(); void cleanTroubleCode();
} }

10
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java
View File

@ -50,16 +50,6 @@ public class SysTroubleServiceImpl extends ServiceImpl<SysTroubleMapper, Couplet
sysTroubleMapper.newFaultData(code); sysTroubleMapper.newFaultData(code);
} }
/**
* 添加异常信息
* @param coupletTroubleCode
* @return
*/
@Override
public int insertMsgResq(CoupletTroubleCode coupletTroubleCode) {
return sysTroubleMapper.insertMsgResq(coupletTroubleCode);
}
/** /**
* 清理故障异常信息 * 清理故障异常信息
*/ */

17
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/time/Timer.java
View File

@ -6,6 +6,7 @@ import com.couplet.common.domain.request.VehicleListParams;
import com.couplet.common.redis.service.RedisService; import com.couplet.common.redis.service.RedisService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@ -22,7 +23,7 @@ import java.util.List;
public class Timer { public class Timer {
//redis //redis
@Autowired @Autowired
private RedisService redis; private StringRedisTemplate redis;
//查询车辆列表 //查询车辆列表
@Autowired @Autowired
private VehicleService vehicleService; private VehicleService vehicleService;
@ -35,9 +36,7 @@ public class Timer {
//先查询车辆列表 //先查询车辆列表
List<Vehicle> list = vehicleService.list(new VehicleListParams(null, null, null, null)); List<Vehicle> list = vehicleService.list(new VehicleListParams(null, null, null, null));
for (Vehicle vehicle : list) {
list.forEach(vehicle -> {
//只针对已经上线的车辆 //只针对已经上线的车辆
if (redis.hasKey(vehicle.getVin())) { if (redis.hasKey(vehicle.getVin())) {
@ -48,15 +47,15 @@ public class Timer {
//执行修改下线状态的方法 //执行修改下线状态的方法
Integer i = vehicleService.onOrOutLineByVIN(vehicle.getVin(), 0); Integer i = vehicleService.onOrOutLineByVIN(vehicle.getVin(), 0);
if (0 == 1) { // if (0 == 1) {
log.error("下线状态修改失败"); // log.error("下线状态修改失败");
} // }
log.info("下线状态修改成功"); log.info("下线状态修改成功");
} }
} }
}
});
} }
} }

1
couplet-modules/couplet-business/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports 100644
View File

@ -0,0 +1 @@
com.couplet.business.server.time.Timer

4
couplet-modules/couplet-business/src/main/resources/bootstrap.yml
View File

@ -16,9 +16,11 @@ spring:
discovery: discovery:
# 服务注册地址 # 服务注册地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
namespace: 172469
config: config:
# 配置中心地址 # 配置中心地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
namespace: 172469
# 配置文件格式 # 配置文件格式
file-extension: yml file-extension: yml
# 共享配置 # 共享配置
@ -28,4 +30,4 @@ spring:
allow-bean-definition-overriding: true allow-bean-definition-overriding: true
logging: logging:
level: level:
com.couplet.trouble.mapper: DEBUG com.couplet.business.server.mapper: DEBUG

15
couplet-modules/couplet-business/src/main/resources/mapper/business/SysTroubleMapper.xml
View File

@ -7,18 +7,14 @@
<resultMap type="com.couplet.common.domain.CoupletTroubleCode" id="SysTroubleResult"> <resultMap type="com.couplet.common.domain.CoupletTroubleCode" id="SysTroubleResult">
<id property="troubleId" column="trouble_id"/> <id property="troubleId" column="trouble_id"/>
<result property="troubleCode" column="trouble_code"/> <result property="troubleCode" column="trouble_code"/>
<result property="troubleValue" column="trouble_value"/> <result property="troubleVin" column="trouble_vin"/>
<result property="troublePosition" column="trouble_position"/>
<result property="troubleTag" column="trouble_tag"/> <result property="troubleTag" column="trouble_tag"/>
<result property="typeId" column="type_id"/> <result property="troubleStartTime" column="trouble_start_time"/>
<result property="gradeId" column="grade_id"/> <result property="troubleEndTime" column="trouble_end_time"/>
<result property="troubleType" column="trouble_type"/>
</resultMap> </resultMap>
<sql id="selectTroubleVo"> <sql id="selectTroubleVo">
select t.*,g.grade_name,y.type_name from couplet_trouble_code t select * from couplet_trouble_code
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> </sql>
<insert id="newFaultData"> <insert id="newFaultData">
INSERT INTO `couplet-cloud`.`couplet_trouble_code` INSERT INTO `couplet-cloud`.`couplet_trouble_code`
@ -26,9 +22,6 @@
`trouble_start_time`) `trouble_start_time`)
VALUES VALUES
(#{troubleVin}, #{troubleTag}, #{troubleStartTime}) (#{troubleVin}, #{troubleTag}, #{troubleStartTime})
</insert>
<insert id="insertMsgResq">
</insert> </insert>
<update id="cleanTroubleCode"> <update id="cleanTroubleCode">

83
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/controller/KafkaTest.java
View File

@ -1,54 +1,93 @@
package com.couplet.mq.controller; package com.couplet.mq.controller;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import javax.annotation.PostConstruct;
import java.time.Duration; import java.time.Duration;
import java.util.Collections; import java.util.Collections;
import java.util.Properties; import java.util.Properties;
/** /**
* @author DongXiaoDong * @ProjectName: five-groups-couplet
* @version 1.0 * @Author: LiuYunHu
* @date 2024/4/5 21:38 * @CreateTime: 2024/4/4
* @description * @Description: kafka测试类
*/ */
@Slf4j
public class KafkaTest { public class KafkaTest {
private static final String TOPIC_NAME = "online"; private static final String TOPIC_NAME = "online";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
public static void main(String[] args) { public static void main(String[] args) {
aaa(); //生产者示例
// produceMessage();
//消费者示例
// consumerMessages();
} }
private static void aaa() { //生产者
@PostConstruct
private static void produceMessage() {
Properties props = new Properties(); Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//创建消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
//创建生产者
try { try {
//订阅主题 //发送消息
consumer.subscribe(Collections.singletonList("online")); for (int i = 0; i < 10000; i++) {
String message = "佳佳来喽" + (i + 1);
producer.send(new ProducerRecord<>(TOPIC_NAME, message));
//持续消费消息 System.out.println("发送消息：" + message);
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
records.forEach(record -> {
System.out.println("消费者接受到的消息值：" + record.value());
});
} }
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); e.printStackTrace();
} finally { } finally {
consumer.close(); producer.close();
} }
} }
//消费者
// private static void consumerMessages() {
// Properties props = new Properties();
// props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
// props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
// props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
// props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
//
// //创建消费者
// KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
//
// try {
//
// //订阅主题
// consumer.subscribe(Collections.singletonList(TOPIC_NAME));
//
// //持续消费消息
// while (true) {
// ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
// records.forEach(record -> {
// System.out.println("消费者接受到的消息值：" + record.value());
// });
// }
// } catch (Exception e) {
// e.printStackTrace();
// } finally {
// consumer.close();
// }
// }
} }

1
couplet-modules/couplet-modules-mq/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
View File

@ -1 +1,2 @@
com.couplet.mq.config.RabbitMQConfig com.couplet.mq.config.RabbitMQConfig
com.couplet.mq.remote.factory.RemoteFenceFallbackFactory

170
couplet-modules/couplet-modules-onLine/src/main/java/com/couplet/online/utils/MqttMonitor.java
View File

@ -2,7 +2,6 @@ package com.couplet.online.utils;
import com.couplet.common.domain.Vehicle; import com.couplet.common.domain.Vehicle;
import com.couplet.remote.RemoteVehicleService; import com.couplet.remote.RemoteVehicleService;
import com.fasterxml.jackson.databind.ser.std.StringSerializer;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerConfig;
@ -17,6 +16,8 @@ import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import java.util.List; import java.util.List;
import java.util.Properties; import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
/** /**
@ -78,6 +79,11 @@ public class MqttMonitor {
private static final String TOPIC_NAME = "online"; private static final String TOPIC_NAME = "online";
private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092"; private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
//线程池，用于异步处理消息到来时的业务逻辑
private ExecutorService executorService = Executors.newSingleThreadExecutor();
//Kafka生产者实例化为类成员变量
private KafkaProducer<String, String> kafkaProducer;
//随项目启动而执行这个方法 //随项目启动而执行这个方法
@PostConstruct @PostConstruct
@ -116,7 +122,7 @@ public class MqttMonitor {
client.reconnect(); client.reconnect();
log.info("重连中..."); log.info("重连中...");
} catch (InterruptedException | MqttException e) { } catch (InterruptedException | MqttException e) {
throw new RuntimeException(e); log.error("重连失败：" + e.getMessage());
} }
} }
@ -124,72 +130,7 @@ public class MqttMonitor {
@Override @Override
public void messageArrived(String topic, MqttMessage mqttMessage) { public void messageArrived(String topic, MqttMessage mqttMessage) {
// log.info("消息已送达"); executorService.execute(() -> processMessageArrival(topic, mqttMessage));
// log.info("接收消息主题：{}",topic);
// log.info("接收消息qos：{}", mqttMessage.getQos());
//接收到的原始报文
String message = new String(mqttMessage.getPayload());
// log.info("接收消息原始内容：{}", message);
//解析后的字符串
String parseMsg = ParseMessageUtil.parseMsg(message);
//拿到前17位（车辆vin码）
String start17 = parseMsg.substring(0, 17);
log.info("当前车辆的vin码为：" + start17);
// //判断缓存中是否有这个vin
// if (redis.hasKey("不存在的车辆VIN" + start17)) {
//
// //可使用RabbitMQ发送消息
// log.error("vin码为" + start17 + "的车辆不属于本系统！");
//
// } else {//如果缓存中没有存这个vin
//
//
// }
//调取接口，通过vin查询车辆
List<Vehicle> vehicles = remoteVehicleService.findByVIN(start17).getData();
System.out.println("**************" + vehicles);
//如果不存在这个车
if (0 == vehicles.size()) {
//将不属于自己系统的车辆存入缓存，便于提前进行拒绝提示
// redis.setCacheObject("不存在的车辆VIN" + start17, start17);
log.error("未找到vin码为" + start17 + "的车辆信息");
} else {
//如果存在这个车
Vehicle vehicle = vehicles.get(0);
log.info("远程调用查询到的车辆数据：" + vehicle);
//上线车辆存入redis 6秒 用于判断车辆是否下线，还要写定时器，定时查询
redis.opsForValue().set(start17, start17, 6L, TimeUnit.SECONDS);
log.info("vin码为" + start17 + "的车辆属于本系统，允许上线！");
//调用上线接口，修改上线状态
Integer i = remoteVehicleService.onOrOutLineByVIN(start17 + "," + 1);
//上线成功
if (0 != i) {
log.info("上线成功！");
try {
produceMessage(message);
} catch (Exception e) {
e.printStackTrace();
}
}
}
} }
@Override @Override
@ -203,31 +144,104 @@ public class MqttMonitor {
} catch (MqttException e) { } catch (MqttException e) {
log.error("mqtt监听者启动失败，{}", e.getMessage()); log.error("mqtt监听者启动失败，{}", e.getMessage());
throw new RuntimeException(e); // throw new RuntimeException(e);
} }
} }
//Kafka生产者 //异步处理，mqtt消息到达后的逻辑
private static void produceMessage(String message) { private void processMessageArrival(String topic, MqttMessage mqttMessage) {
// log.info("消息已送达");
// log.info("接收消息主题：{}",topic);
// log.info("接收消息qos：{}", mqttMessage.getQos());
//接收到的原始报文
String message = new String(mqttMessage.getPayload());
//log.info("接收消息原始内容：{}", message);
//解析后的字符串
String parseMsg = ParseMessageUtil.parseMsg(message);
//拿到前17位（车辆vin码）
String start17 = parseMsg.substring(0, 17);
log.info("当前车辆的vin码为：" + start17);
// //判断缓存中是否有这个vin
// if (redis.hasKey("不存在的车辆VIN" + start17)) {
//
// //可使用RabbitMQ发送消息
// log.error("vin码为" + start17 + "的车辆不属于本系统！");
//
// } else {//如果缓存中没有存这个vin
//
//
// }
//调取接口，通过vin查询车辆
List<Vehicle> vehicles = remoteVehicleService.findByVIN(start17).getData();
System.out.println("**************" + vehicles);
//如果不存在这个车
if (0 == vehicles.size()) {
//将不属于自己系统的车辆存入缓存，便于提前进行拒绝提示
// redis.setCacheObject("不存在的车辆VIN" + start17, start17);
log.error("未找到vin码为" + start17 + "的车辆信息");
} else {
//如果存在这个车
Vehicle vehicle = vehicles.get(0);
log.info("远程调用查询到的车辆数据：" + vehicle);
//上线车辆存入redis 6秒 用于判断车辆是否下线，还要写定时器，定时查询
redis.opsForValue().set(start17, start17, 6L, TimeUnit.SECONDS);
log.info("vin码为" + start17 + "的车辆属于本系统，允许上线！");
//调用上线接口，修改上线状态
Integer i = remoteVehicleService.onOrOutLineByVIN(start17 + "," + 1);
//上线成功
if (0 != i) {
log.info("上线成功！");
try {
produceMessage(message);
} catch (Exception e) {
log.error("发送信息异常：" + e.getMessage());
}
}
}
}
//Kafka生产者实例化为类成员变量
@PostConstruct
public void initKafkaProducer() {
Properties props = new Properties(); Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(props); kafkaProducer = new KafkaProducer<>(props);
//创建生产者 }
try {
//发送消息至Kafka
private void produceMessage(String message) {
try {
//发送消息 //发送消息
producer.send(new ProducerRecord<>(TOPIC_NAME, message)); kafkaProducer.send(new ProducerRecord<>(TOPIC_NAME, message));
System.out.println("发送消息：" + message); System.out.println("发送消息：" + message);
} catch (Exception e) { } catch (Exception e) {
e.printStackTrace(); log.error("消息发送失败：" + e.getMessage());
} finally {
producer.close();
} }
} }
} }

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

@ -15,9 +15,11 @@ spring:
discovery: discovery:
# 服务注册地址 # 服务注册地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
namespace: 172469
config: config:
# 配置中心地址 # 配置中心地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
namespace: 172469
# 配置文件格式 # 配置文件格式
file-extension: yml file-extension: yml
# 共享配置 # 共享配置

0
mqttx_32dcaf76-tcp8130181161883/.lck 100644
View File