six-vehicle
/
cloud-server
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev.analysis' into dev

dev
LQS 2024-10-12 11:13:58 +08:00
parent 91cb9ab991 1631a7cae8
commit 7c67737481
8 changed files with 403 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
JavaSample-tcp1061513671883/.lck 100644
View File

47
cloud-modules/cloud-modules-enterprise/cloud-modules-enterprise-common/src/main/java/com/muyu/enterprise/domain/MqttProperties.java 100644
View File

@ -0,0 +1,47 @@
package com.muyu.enterprise.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
*
* @Author：李庆帅
* @Package：com.muyu.enterprise.domain
* @Project：cloud-server
* @name：MqttProperties
* @Date：2024/10/10 20:01
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class MqttProperties
{
/**
* 节点
*/
private String broker;
/**
* 主题
*/
private String topic;
/**
* 用户名
*/
private String userName;
/**
* 密码
*/
private String password;
/**
* 客户端ID
*/
private String clientId;
/**
* 上报级别
*/
private int qos=0;
}

11
cloud-modules/cloud-modules-protocol-analysis/pom.xml
View File

@ -136,6 +136,17 @@
<artifactId>cloud-modules-enterprise-cache</artifactId>
</dependency>
<!-- rabbit消息队列模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-rabbit</artifactId>
</dependency>
<!-- &lt;!&ndash; gateway系统模块 &ndash;&gt;-->
<!-- <dependency>-->
<!-- <groupId>com.muyu</groupId>-->
<!-- <artifactId>cloud-modules-vehicle-gateway</artifactId>-->
<!-- </dependency>-->
</dependencies>

189
cloud-modules/cloud-modules-protocol-analysis/src/main/java/com/muyu/analysis/parsing/MQTT/ParsingMQTT.java
View File

@ -1,5 +1,6 @@
package com.muyu.analysis.parsing.mqtt;
import com.muyu.analysis.parsing.mqtt.service.MqttClientService;
import com.muyu.common.core.constant.KafkaConstants;
import com.muyu.common.core.constant.RedisConstants;
import com.muyu.common.core.domain.Result;
@ -7,6 +8,7 @@ import com.muyu.common.mqtt.MQTTConnect;
import com.muyu.enterprise.cache.CarMessageValueCacheService;
import com.muyu.enterprise.cache.CarVehicleCacheService;
import com.muyu.enterprise.cache.CarVehicleTypeCacheService;
import com.muyu.enterprise.domain.MqttProperties;
import com.muyu.enterprise.domain.car.Vehicle;
import com.muyu.enterprise.domain.car.VehicleType;
import com.muyu.enterprise.domain.resp.car.MessageValueListResp;
@ -20,6 +22,7 @@ import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
@ -64,109 +67,121 @@ public class ParsingMQTT {
@Resource
private CarMessageValueCacheService allMessageValueCacheService;
@Autowired
private MqttClientService mqttClientService;
/**
* 协议解析
*/
@PostConstruct
public void mqttClient() {
public JSONObject mqttClient(MqttMessage mqttMessage,String messageStr) {
// String topic = "vehicle";
//// String MQTTConnect.BROKER = "tcp://106.15.136.7:1883";
//// String clientId = "JavaSample";
try {
// 第三个参数为空，默认持久化策略
MqttClient sampleClient = new MqttClient(MQTTConnect.BROKER, MQTTConnect.CLIENT_ID);
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(true);
log.info("连接中MQTTConnect.BROKER: {}", MQTTConnect.BROKER);
sampleClient.connect(connOpts);
sampleClient.subscribe(MQTTConnect.TOPIC, 0);
sampleClient.setCallback(new MqttCallback() {
// 连接丢失
@Override
public void connectionLost(Throwable throwable) {
String payload = new String(mqttMessage.getPayload());
log.info("====:{}", payload);
}
// try {
// // 第三个参数为空，默认持久化策略
// MqttClient sampleClient = new MqttClient(MQTTConnect.BROKER, MQTTConnect.CLIENT_ID);
// MqttConnectOptions connOpts = new MqttConnectOptions();
// connOpts.setCleanSession(true);
// log.info("连接中MQTTConnect.BROKER: {}", MQTTConnect.BROKER);
// sampleClient.connect(connOpts);
// //MQTTConnect.TOPIC
// sampleClient.subscribe(MQTTConnect.TOPIC, 0);
// sampleClient.setCallback(new MqttCallback() {
// // 连接丢失
// @Override
// public void connectionLost(Throwable throwable) {
//
// }
//
// // 连接成功
// @Override
// public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
// System.out.println(new String(mqttMessage.getPayload()));
//// JSONObject entries = ParsingMQTT.protocolParsing(new String(mqttMessage.getPayload()));
////
//// ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.MESSAGE_PARSING,
//// entries.toString() );
//// kafkaProducer.send(producerRecord);
//// log.info("解析之后的数据："+entries);
//
// }
//
//
//
// // 接收信息
// @Override
// public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
//
// }
// });
// } catch (MqttException me) {
// log.info("reason " + me.getReasonCode());
// log.info("msg " + me.getMessage());
// log.info("loc " + me.getLocalizedMessage());
// log.info("cause " + me.getCause());
// log.info("excep " + me);
// System.out.println("reason " + me.getReasonCode());
// System.out.println("msg " + me.getMessage());
// System.out.println("loc " + me.getLocalizedMessage());
// System.out.println("cause " + me.getCause());
// System.out.println("excep " + me);
// me.printStackTrace();
// }
// 连接成功
@Override
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
System.out.println(new String(mqttMessage.getPayload()));
JSONObject entries = this.protocolParsing(new String(mqttMessage.getPayload()));
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.MESSAGE_PARSING,
entries.toString() );
kafkaProducer.send(producerRecord);
log.info("解析之后的数据："+entries);
}
/**
* 协议解析
* @param messageStr
* @return
*/
public JSONObject protocolParsing(String messageStr) {
//根据空格切割数据
String[] hexArray = messageStr.split(" ");
StringBuilder result = new StringBuilder();
//遍历十六进制数据转换为字符
for (String hex : hexArray) {
int decimal = Integer.parseInt(hex, 16);
result.append((char) decimal);
}
//取出车辆VIN码
String vehicleVin = result.substring(1, 18);
log.info("车辆VIN码: " + vehicleVin);
//根据车辆VIN码查询报文模板ID
// 根据车辆VIN码查询车辆信息
Vehicle vehicle = vehicleCacheService.get(vehicleVin);
VehicleType vehicleType = vehicleTypeCacheService.get(String.valueOf(vehicle.getVehicleTypeId()));
Long templateId = vehicleType.getMessageTemplateId();
List<MessageValueListResp> templateList = allMessageValueCacheService.get(String.valueOf(templateId));
//判断报文模板列表不为空
if (templateList.isEmpty()) {
log.info("报文模版为空");
throw new RuntimeException("报文模版为空");
}
//存储报文模版解析后的数据
JSONObject jsonObject = new JSONObject();
for (MessageValueListResp messageValue : templateList) {
//起始位下标
Integer startIndex = messageValue.getMessageStartIndex() - 1;
//结束位下标
Integer endIndex = messageValue.getMessageEndIndex();
//根据报文模版截取数据
String value = result.substring(startIndex, endIndex);
//存入数据
jsonObject.put(messageValue.getMessageLabel(), value);
}
return jsonObject;
}
// 接收信息
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
});
} catch (MqttException me) {
log.info("reason " + me.getReasonCode());
log.info("msg " + me.getMessage());
log.info("loc " + me.getLocalizedMessage());
log.info("cause " + me.getCause());
log.info("excep " + me);
System.out.println("reason " + me.getReasonCode());
System.out.println("msg " + me.getMessage());
System.out.println("loc " + me.getLocalizedMessage());
System.out.println("cause " + me.getCause());
System.out.println("excep " + me);
me.printStackTrace();
}
/**
* 协议解析
* @param messageStr
* @return
*/
// public JSONObject protocolParsing() {
//根据空格切割数据
String[] hexArray = messageStr.split(" ");
StringBuilder result = new StringBuilder();
//遍历十六进制数据转换为字符
for (String hex : hexArray) {
int decimal = Integer.parseInt(hex, 16);
result.append((char) decimal);
}
//取出车辆VIN码
String vehicleVin = result.substring(1, 18);
log.info("车辆VIN码: " + vehicleVin);
//根据车辆VIN码查询报文模板ID
// 根据车辆VIN码查询车辆信息
Vehicle vehicle = vehicleCacheService.get(vehicleVin);
VehicleType vehicleType = vehicleTypeCacheService.get(String.valueOf(vehicle.getVehicleTypeId()));
Long templateId = vehicleType.getMessageTemplateId();
List<MessageValueListResp> templateList = allMessageValueCacheService.get(String.valueOf(templateId));
//判断报文模板列表不为空
if (templateList.isEmpty()) {
log.info("报文模版为空");
throw new RuntimeException("报文模版为空");
}
//存储报文模版解析后的数据
JSONObject jsonObject = new JSONObject();
for (MessageValueListResp messageValue : templateList) {
//起始位下标
Integer startIndex = messageValue.getMessageStartIndex() - 1;
//结束位下标
Integer endIndex = messageValue.getMessageEndIndex();
//根据报文模版截取数据
String value = result.substring(startIndex, endIndex);
//存入数据
jsonObject.put(messageValue.getMessageLabel(), value);
}
return jsonObject;
// }
}
}

110
cloud-modules/cloud-modules-protocol-analysis/src/main/java/com/muyu/analysis/parsing/MQTT/service/MqttClientService.java 100644
View File

@ -0,0 +1,110 @@
package com.muyu.analysis.parsing.mqtt.service;
import cn.hutool.json.JSONObject;
import com.muyu.analysis.parsing.mqtt.ParsingMQTT;
import com.muyu.common.core.constant.KafkaConstants;
import com.muyu.enterprise.domain.MqttProperties;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @Author：李庆帅
* @Package：com.muyu.analysis.parsing.mqtt.service
* @Project：cloud-server
* @name：MqttClientService
* @Date：2024/10/10 20:24
*/
@Service
@Slf4j
public class MqttClientService {
@Autowired
private ParsingMQTT messageProcessor;
@Resource
private KafkaProducer<String, String> kafkaProducer;
private final ExecutorService executorService = Executors.newCachedThreadPool();
private MqttClient sampleClient;
public void connectAndSubscribeAsync(MqttProperties mqttProperties) {
executorService.submit(() -> {
try {
connectAndSubscribe(mqttProperties);
} catch (MqttException | IOException e) {
log.error("MQTT连接或订阅失败", e);
}
});
}
private void connectAndSubscribe(MqttProperties mqttProperties) throws MqttException, IOException {
if (sampleClient != null && sampleClient.isConnected()) {
log.info("MQTT客户端已经连接，跳过重新连接。");
return;
}
sampleClient = new MqttClient(mqttProperties.getBroker(), mqttProperties.getClientId());
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setUserName(mqttProperties.getUserName());
connOpts.setPassword(mqttProperties.getPassword().toCharArray());
connOpts.setCleanSession(true);
sampleClient.connect(connOpts);
sampleClient.subscribe(mqttProperties.getTopic(), 0);
sampleClient.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable throwable) {
log.error("连接丢失：{}", throwable.getMessage());
}
@Override
public void messageArrived(String s, MqttMessage mqttMessage) {
executorService.submit(() -> messageProcessor.mqttClient(mqttMessage,new String(mqttMessage.getPayload())));
System.out.println(new String(mqttMessage.getPayload()));
JSONObject entries =messageProcessor.mqttClient(mqttMessage,new String(mqttMessage.getPayload()));
// JSONObject entries = o.protocolParsing(new String(mqttMessage.getPayload()));
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.MESSAGE_PARSING,
entries.toString() );
kafkaProducer.send(producerRecord);
log.info("解析之后的数据："+entries);
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("消息发送完成：{}", iMqttDeliveryToken);
}
});
}
@PreDestroy
public void shutdown() {
executorService.shutdown();
if (sampleClient != null && sampleClient.isConnected()) {
try {
sampleClient.disconnect();
} catch (MqttException e) {
log.error("MQTT客户端断开连接失败", e);
}
}
}
}

43
cloud-modules/cloud-modules-protocol-analysis/src/main/java/com/muyu/analysis/parsing/consumer/RabbitListenerComponent.java 100644
View File

@ -0,0 +1,43 @@
package com.muyu.analysis.parsing.consumer;
import com.muyu.analysis.parsing.mqtt.service.MqttClientService;
import com.muyu.enterprise.domain.MqttProperties;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @Author：李庆帅
* @Package：com.muyu.analysis.parsing.consumer
* @Project：cloud-server
* @name：RabbitListenerComponent
* @Date：2024/10/10 20:24
*/
@Component
@Slf4j
public class RabbitListenerComponent {
@Autowired
private MqttClientService mqttClientService;
private static final String FORM_QUEUE = "GO_LINE";
@RabbitListener(queuesToDeclare = @Queue(value = FORM_QUEUE, durable = "true"))
public void downline(MqttProperties mqttProperties, Message message, Channel channel) {
try {
mqttClientService.connectAndSubscribeAsync(mqttProperties);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
e.printStackTrace();
log.error("处理RabbitMQ消息时发生错误", e);
}
}
}

83
cloud-modules/cloud-modules-protocol-analysis/src/main/java/com/muyu/analysis/parsing/manager/MessageProcessor.java 100644
View File

@ -0,0 +1,83 @@
//package com.muyu.analysis.parsing.manager;
//
//
//import com.alibaba.fastjson2.JSONObject;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.producer.KafkaProducer;
//import org.apache.kafka.clients.producer.ProducerRecord;
//import org.eclipse.paho.client.mqttv3.MqttMessage;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.Resource;
//import java.util.ArrayList;
//import java.util.List;
//
//
//
///**
// *
// *
// * @Author：李庆帅
// * @Package：com.muyu.analysis.parsing.manager
// * @Project：cloud-server
// * @name：MessageProcessor
// * @Date：2024/10/10 20:24
// */
//
//@Slf4j
//@Service
//public class MessageProcessor {
// private static final int ID = 1;
// @Resource
// private SysCarMessageServiceImpl sysCarMessageService;
// @Resource
// private KafkaProducer<String, String> kafkaProducer;
//
//
// public void processMessage(MqttMessage mqttMessage) {
// String payload = new String(mqttMessage.getPayload());
// log.info("====:{}", payload);
//
// List<SysCarMessage> carMessages = sysCarMessageService.selectSysCarMessageLists(ID);
// List<KafKaData> kafKaDataList = new ArrayList<>();
//
// String[] test = payload.split(" ");
// for (SysCarMessage carMessage : carMessages) {
// int start = Integer.parseInt(carMessage.getMessageStartIndex()) - 1;
// int end = Integer.parseInt(carMessage.getMessageEndIndex());
//
// StringBuilder hexBuilder = new StringBuilder();
// for (int i = start; i < end; i++) {
// hexBuilder.append(test[i]);
// }
//
// String hex = hexBuilder.toString();
// char[] result = new char[hex.length() / 2];
// for (int x = 0; x < hex.length(); x += 2) {
// int high = Character.digit(hex.charAt(x), 16);
// int low = Character.digit(hex.charAt(x + 1), 16);
// result[x / 2] = (char) ((high << 4) + low);
// }
//
// String value = new String(result);
// kafKaDataList.add(KafKaData.builder()
// .key(carMessage.getMessageTypeCode())
// .label(carMessage.getMessageTypeCode())
// .value(value)
// .type(carMessage.getMessageType())
// .build());
// }
//
// kafKaDataList.add(KafKaData.builder()
// .key("firmCode")
// .label("企业编码")
// .value("firm01")
// .type("String")
// .build());
// String jsonString = JSONObject.toJSONString(kafKaDataList);
// ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.KafkaTopic, jsonString);
// kafkaProducer.send(producerRecord);
// log.info("kafka投产：{}", jsonString);
// }
//
//}

7
pom.xml
View File

@ -425,6 +425,13 @@
<version>${muyu.version}</version>
</dependency>
<!-- gateway系统模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-vehicle-gateway</artifactId>
<version>${muyu.version}</version>
</dependency>
</dependencies>
</dependencyManagement>