yaoxin-jiangpeng
/
mqtt-message
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: 新增mqtt客户端监听服务器节点,获取报文并解析

master
yaoxin 2024-06-05 22:24:22 +08:00
parent e5c26727ac
commit 8382839122
6 changed files with 515 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
src/main/java/com/muyu/mqttmessage/common/MqttMessageModel.java 100644
View File

@ -0,0 +1,36 @@
package com.muyu.mqttmessage.common;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.util.UUID;
/**
* @ClassName MqttMessageInfo
* @Description 描述
* @Author Xin.Yao
* @Date 2024/5/30 上午8:41
*/
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
public class MqttMessageModel {
private String broker;
private String topic;
private String username;
private String password;
private String clientId;
public static MqttMessageModel builderMqttMessage(String broker, String topic,String username,String password) {
return MqttMessageModel.builder()
.broker(broker)
.topic(topic)
.clientId(UUID.randomUUID().toString())
.username(username)
.password(password)
.build();
}
}

64
src/main/java/com/muyu/mqttmessage/config/MqttFactory.java
View File

@ -1,11 +1,9 @@
package com.muyu.mqttmessage.config;
import com.muyu.loadcenter.common.MqttMessageModel;
import com.muyu.loadcenter.service.impl.MqttCallBackServiceImpl;
import com.muyu.mqttmessage.common.MqttMessageModel;
import com.muyu.mqttmessage.service.impl.MqttCallBackServiceImpl;
import lombok.AllArgsConstructor;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;
@ -19,19 +17,67 @@ import org.springframework.stereotype.Service;
@AllArgsConstructor
public class MqttFactory {
private final MqttCallBackServiceImpl mqttCallBackService;
public MqttClient createMqttClient(MqttMessageModel mqttMessageModel) {
private static MqttCallBackServiceImpl mqttCallBackService;
// public static void main(String[] args){
// String broker = "tcp://43.142.44.217:1883";
// String topic = "mqtt001";
// String username = "emqx";
// String password = "public";
// String clientid = "subscribe_client";
// int qos = 0;
//
// try {
// MqttClient client = new MqttClient(broker, clientid, new MemoryPersistence());
// // 连接参数
// MqttConnectOptions options = new MqttConnectOptions();
// options.setUserName(username);
// options.setPassword(password.toCharArray());
// options.setConnectionTimeout(60);
// options.setKeepAliveInterval(60);
// // 设置回调
// client.setCallback(new MqttCallback() {
//
// public void connectionLost(Throwable cause) {
// System.out.println("connectionLost: " + cause.getMessage());
// }
//
// public void messageArrived(String topic, MqttMessage message) {
// System.out.println("topic: " + topic);
// System.out.println("Qos: " + message.getQos());
// System.out.println("message content: " + new String(message.getPayload()));
//
// }
//
// public void deliveryComplete(IMqttDeliveryToken token) {
// System.out.println("deliveryComplete---------" + token.isComplete());
// }
//
// });
// client.connect(options);
// client.subscribe(topic, qos);
// } catch (Exception e) {
// e.printStackTrace();
// }
// }
public static void main(String[] args) {
MqttMessageModel mqttMessageModel1 = MqttMessageModel.builderMqttMessage("tcp://43.142.44.217:1883", "mqtt001","1111","22222");
MqttFactory.createMqttClient(mqttMessageModel1);
MqttMessageModel mqttMessageModel2 = MqttMessageModel.builderMqttMessage("tcp://47.98.170.220:1883", "mqtt002","1111","22222");
MqttFactory.createMqttClient(mqttMessageModel2);
}
public static MqttClient createMqttClient(MqttMessageModel mqttMessageModel) {
MqttClient client =null;
int qos = 0;
try {
client = new MqttClient(mqttMessageModel.getBroker(), mqttMessageModel.getClientid(), new MemoryPersistence());
client = new MqttClient(mqttMessageModel.getBroker(), mqttMessageModel.getClientId(), new MemoryPersistence());
// 连接参数
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName(mqttMessageModel.getUsername());
options.setPassword(mqttMessageModel.getPassword().toCharArray());
options.setConnectionTimeout(60);
options.setKeepAliveInterval(60);
client.setCallback(mqttCallBackService);
client.setCallback(new MqttCallBackServiceImpl());
client.connect(options);
client.subscribe(mqttMessageModel.getTopic(), qos);
} catch (MqttException e) {

261
src/main/java/com/muyu/mqttmessage/domain/VehicleData.java 100644
View File

@ -0,0 +1,261 @@
package com.muyu.mqttmessage.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.math.BigDecimal;
/**
* @ClassName VehicleData
* @Description 描述
* @Author Xin.Yao
* @Date 2024/6/5 下午6:52
*/
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
public class VehicleData {
/**
* VIN
*/
private String vin;
/**
* 行驶路线
*/
private String drivingRoute;
/**
* 经度
*/
private String longitude;
/**
* 维度
*/
private String latitude;
/**
* 速度
*/
private String speed;
/**
* 里程
*/
private BigDecimal mileage;
/**
* 总电压
*/
private String voltage;
/**
* 总电流
*/
private String current;
/**
* 绝缘电阻
*/
private String resistance;
/**
* 档位
*/
private String gear = "P";
/**
* 加速踏板行程值
*/
private String accelerationPedal;
/**
* 制动踏板行程值
*/
private String brakePedal;
/**
* 燃料消耗率
*/
private String fuelConsumptionRate;
/**
* 电机控制器温度
*/
private String motorControllerTemperature;
/**
* 电机转速
*/
private String motorSpeed;
/**
* 电机转矩
*/
private String motorTorque;
/**
* 电机温度
*/
private String motorTemperature;
/**
* 电机电压
*/
private String motorVoltage;
/**
* 电机电流
*/
private String motorCurrent;
/**
* 动力电池剩余电量SOC
*/
private BigDecimal remainingBattery;
/**
* 电池总容量
*/
private BigDecimal batteryLevel;
/**
* 当前状态允许的最大反馈功率
*/
private String maximumFeedbackPower;
/**
* 当前状态允许最大放电功率
*/
private String maximumDischargePower;
/**
* BMS自检计数器
*/
private String selfCheckCounter;
/**
* 动力电池充放电电流
*/
private String totalBatteryCurrent;
/**
* 动力电池负载端总电压V3
*/
private String totalBatteryVoltage;
/**
* 单次最大电压
*/
private String singleBatteryMaxVoltage;
/**
* 单体电池最低电压
*/
private String singleBatteryMinVoltage;
/**
* 单体电池最高温度
*/
private String singleBatteryMaxTemperature;
/**
* 单体电池最低温度
*/
private String singleBatteryMinTemperature;
/**
* 动力电池可用容量
*/
private String availableBatteryCapacity;
/**
* 车辆状态
*/
private int vehicleStatus = 1;
/**
* 充电状态
*/
private int chargingStatus = 1;
/**
* 运行状态
*/
private int operatingStatus = 1;
/**
* SOC
*/
private int socStatus = 1;
/**
* 可充电储能装置工作状态
*/
private int chargingEnergyStorageStatus = 1;
/**
* 驱动电机状态
*/
private int driveMotorStatus = 1;
/**
* 定位是否有效
*/
private int positionStatus = 1;
/**
* EAS(汽车防盗系统)状态
*/
private int easStatus = 1;
/**
* PTC(电动加热器)状态
*/
private int ptcStatus = 1;
/**
* EPS(电动助力系统)状态
*/
private int epsStatus = 1;
/**
* ABS(防抱死)状态
*/
private int absStatus = 1;
/**
* MCU(电机/逆变器)状态
*/
private int mcuStatus = 1;
/**
* 动力电池加热状态
*/
private int heatingStatus = 1;
/**
* 动力电池当前状态
*/
private int batteryStatus = 1;
/**
* 动力电池保温状态
*/
private int batteryInsulationStatus = 1;
/**
* DCDC(电力交换系统)状态
*/
private int dcdcStatus = 1;
/**
* CHG(充电机)状态
*/
private int chgStatus = 1;
}

125
src/main/java/com/muyu/mqttmessage/service/impl/MqttCallBackServiceImpl.java
View File

@ -1,10 +1,16 @@
package com.muyu.mqttmessage.service.impl;
import com.alibaba.fastjson2.JSON;
import com.muyu.mqttmessage.domain.VehicleData;
import com.muyu.mqttmessage.utils.ConversionUtil;
import lombok.extern.log4j.Log4j2;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;
import java.math.BigDecimal;
/**
* @ClassName MqttCallBackConfig
* @Description 描述
@ -12,6 +18,7 @@ import org.springframework.stereotype.Component;
* @Date 2024/5/30 上午8:36
*/
@Component
@Log4j2
public class MqttCallBackServiceImpl implements MqttCallback {
@Override
public void connectionLost(Throwable cause) {
@ -20,13 +27,125 @@ public class MqttCallBackServiceImpl implements MqttCallback {
@Override
public void messageArrived(String topic, MqttMessage message) {
System.out.println("topic: " + topic);
System.out.println("Qos: " + message.getQos());
System.out.println("message content: " + new String(message.getPayload()));
log.info("服务器{}监听的报文: {}" ,topic, ConversionUtil.hexStringToString(new String(message.getPayload())));
log.info("转化对象:{}", JSON.toJSONString(getVehicleData(ConversionUtil.hexStringToString(new String(message.getPayload())))));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("deliveryComplete---------" + token.isComplete());
}
public VehicleData getVehicleData(String message) {
message = message.substring(1,message.length()-2);
return VehicleData.builder()
//17
.vin(message.substring(0,17))
// 当前时间戳 13
.drivingRoute(message.substring(17,30))
//第二位经度 longitude latitude 11
.longitude(message.substring(30,41))
//第三位维度 longitude latitude 10
.latitude(message.substring(41,51))
//车速 6
.speed(removeSuperfluousDigit(message.substring(51,57)))
//总里程 11
.mileage(BigDecimal.valueOf(Double.valueOf(message.substring(57,68))))
// 总电压 6
.voltage(removeSuperfluousDigit(message.substring(68,74)))
//总电流 5
.current(removeSuperfluousDigit(message.substring(74,79)))
//绝缘电阻 79 - 87 9
.resistance(removeSuperfluousDigit(message.substring(79,88)))
//档位
.gear(message.substring(88,89))
// 加速踏板行程值 2
.accelerationPedal(removeSuperfluousDigit(message.substring(89,91)))
// 制动踏板行程值 2
.brakePedal(removeSuperfluousDigit(message.substring(91,93)))
// 燃料消耗率 5
.fuelConsumptionRate(removeSuperfluousDigit(message.substring(93,98)))
//电机控制器温度 6
.motorControllerTemperature(removeSuperfluousDigit(message.substring(98,104)))
//电机转速 5
.motorSpeed(removeSuperfluousDigit(message.substring(104,109)))
//点击转矩 4
.motorTorque(removeSuperfluousDigit(message.substring(109,113)))
//电机温度 6
.motorTemperature(removeSuperfluousDigit(message.substring(113,119)))
//电机电压 5
.motorVoltage(removeSuperfluousDigit(message.substring(119,124)))
//电机电流 8
.motorCurrent(removeSuperfluousDigit(message.substring(124,132)))
//动力电池剩余电量SOC 6
.remainingBattery(BigDecimal.valueOf(Double.valueOf(message.substring(132,138))))
//当前状态允许的最大反馈功率 6
.maximumFeedbackPower(removeSuperfluousDigit(message.substring(138,144)))
//当前状态允许最大放电功率 6
.maximumDischargePower(removeSuperfluousDigit(message.substring(144,150)))
//BMS自检计数器 2
.selfCheckCounter(removeSuperfluousDigit(message.substring(150,152)))
//动力电池充放电电流 5
.totalBatteryCurrent(removeSuperfluousDigit(message.substring(152,157)))
//动力电池负载端总电压V3 6
.totalBatteryVoltage(removeSuperfluousDigit(message.substring(157,163)))
//单次最大电压 4
.singleBatteryMaxVoltage(removeSuperfluousDigit(message.substring(163,167)))
//单体电池最低电压 4
.singleBatteryMinVoltage(removeSuperfluousDigit(message.substring(167,171)))
//单体电池最高温度 4
.singleBatteryMaxTemperature(removeSuperfluousDigit(message.substring(171,177)))
//单体电池最低温度 6
.singleBatteryMinTemperature(removeSuperfluousDigit(message.substring(177,183)))
//动力电池可用容量 6
.availableBatteryCapacity(removeSuperfluousDigit(message.substring(183,189)))
//车辆状态
.vehicleStatus(Integer.valueOf(message.substring(189,190)))
//充电状态
.chargingStatus(Integer.valueOf(message.substring(190,191)))
//运行状态
.operatingStatus(Integer.valueOf(message.substring(191,192)))
//SOC
.socStatus(Integer.valueOf(message.substring(192,193)))
//可充电储能装置工作状态
.chargingEnergyStorageStatus(Integer.valueOf(message.substring(193,194)))
//驱动电机状态
.driveMotorStatus(Integer.valueOf(message.substring(194,195)))
//定位是否有效
.positionStatus(Integer.valueOf(message.substring(195,196)))
//EAS
.easStatus(Integer.valueOf(message.substring(196,197)))
//PTC
.ptcStatus(Integer.valueOf(message.substring(197,198)))
//EPS
.epsStatus(Integer.valueOf(message.substring(198,199)))
//ABS
.absStatus(Integer.valueOf(message.substring(199,200)))
//MCU
.mcuStatus(Integer.valueOf(message.substring(200,201)))
//动力电池加热状态
.heatingStatus(Integer.valueOf(message.substring(201,202)))
//动力电池当前状态
.batteryStatus(Integer.valueOf(message.substring(202,203)))
//动力电池保温状态
.batteryInsulationStatus(Integer.valueOf(message.substring(203,204)))
//DCDC
.dcdcStatus(Integer.valueOf(message.substring(204,205)))
//CHG
.chgStatus(Integer.valueOf(message.substring(205,206)))
.build();
}
public String removeSuperfluousDigit(String str){
if(str.length()>1){
if(str.charAt(0)=='0'){
return removeSuperfluousDigit(str.substring(1));
}else{
return str;
}
}else{
return str;
}
}
}

41
src/main/java/com/muyu/mqttmessage/utils/ConversionUtil.java 100644
View File

@ -0,0 +1,41 @@
package com.muyu.mqttmessage.utils;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
/**
* @ClassName ConversionUtil
* @Description 解析报文类
* @Author Xin.Yao
* @Date 2024/6/5 下午6:32
*/
@Component
public class ConversionUtil {
/**
* 将16进制转化为字符串
* @param hexString
* @return hexString
*/
public static String hexStringToString(String hexString){
if (hexString == null || "".equals(hexString)){
return null;
}
hexString = hexString.replace(" ","");
byte[] bytes = new byte[hexString.length() / 2];
for (int i = 0; i < bytes.length; i++){
try {
bytes[i] = (byte) (0xff & Integer.parseInt(hexString.substring(i * 2, i * 2 + 2), 16));
}catch (Exception e){
e.printStackTrace();
}
}
try{
hexString = new String(bytes, StandardCharsets.UTF_8);
}catch (Exception e){
e.printStackTrace();
}
return hexString;
}
}

13
src/test/java/com/muyu/mqttmessage/MqttMessageApplicationTests.java
View File

@ -1,13 +0,0 @@
package com.muyu.mqttmessage;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class MqttMessageApplicationTests {
@Test
void contextLoads() {
}
}