dyf/fys-Multi-tenant
1
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

设备mqtt收发数据

Browse Source
This commit is contained in:
chenyouting
2025-07-30 08:50:44 +08:00
parent 2b2edf096d
commit ac353b1078
21 changed files with 679 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/config/MqttConfiguration.java
View File

@ -1,31 +0,0 @@
package com.fuyuanshen.system.mqtt.config;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
@Configuration
public class MqttConfiguration {
@Autowired
private MqttPropertiesConfig mqttPropertiesConfig;
/** 创建连接工厂 **/
@Bean
public MqttPahoClientFactory mqttPahoClientFactory(){
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true); //设置新会话
options.setUserName(mqttPropertiesConfig.getUsername());
options.setPassword(mqttPropertiesConfig.getPassword().toCharArray());
options.setServerURIs(new String[]{mqttPropertiesConfig.getUrl()});
options.setAutomaticReconnect(true); // 启用自动重连
options.setConnectionTimeout(10); // 设置连接超时时间
options.setKeepAliveInterval(60); // 设置心跳间隔
factory.setConnectionOptions(options);
return factory;
}
}

17
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/config/MqttGateway.java
View File

@ -1,17 +0,0 @@
package com.fuyuanshen.system.mqtt.config;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
/**
* @Author: HarryLin
* @Date: 2025/3/20 17:06
* @Company: 北京红山信息科技研究院有限公司
* @Email: linyun@***.com.cn
**/
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
public abstract void sendMsgToMqtt(@Header(value = MqttHeaders.TOPIC) String topic, String payload);
public abstract void sendMsgToMqtt(@Header(value = MqttHeaders.TOPIC) String topic, @Header(value = MqttHeaders.QOS) int qos, String payload );
}

58
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/config/MqttInboundConfiguration.java
View File

@ -1,58 +0,0 @@
package com.fuyuanshen.system.mqtt.config;
import cn.hutool.core.lang.UUID;
import com.fuyuanshen.system.mqtt.receiver.ReceiverMessageHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@Configuration
public class MqttInboundConfiguration {
@Autowired
private MqttPropertiesConfig mqttPropertiesConfig;
@Autowired
private MqttPahoClientFactory mqttPahoClientFactory;
@Autowired
private ReceiverMessageHandler receiverMessageHandler;
//消息通道
@Bean
public MessageChannel messageInboundChannel(){
return new DirectChannel();
}
/**
* 配置入站适配器
* 作用： 设置订阅主题，以及指定消息的通道 等相关属性
* */
@Bean
public MessageProducer messageProducer(){
// 生成一个不重复的随机数
String clientId = mqttPropertiesConfig.getSubClientId() + "_" + UUID.fastUUID();
MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter = new MqttPahoMessageDrivenChannelAdapter(
mqttPropertiesConfig.getUrl(),
clientId,
mqttPahoClientFactory,
mqttPropertiesConfig.getSubTopic().split(",")
);
mqttPahoMessageDrivenChannelAdapter.setQos(1);
mqttPahoMessageDrivenChannelAdapter.setConverter(new DefaultPahoMessageConverter());
mqttPahoMessageDrivenChannelAdapter.setOutputChannel(messageInboundChannel());
return mqttPahoMessageDrivenChannelAdapter;
}
/** 指定处理消息来自哪个通道 */
@Bean
@ServiceActivator(inputChannel = "messageInboundChannel")
public MessageHandler messageHandler(){
return receiverMessageHandler;
}
}

46
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/config/MqttOutboundConfiguration.java
View File

@ -1,46 +0,0 @@
package com.fuyuanshen.system.mqtt.config;
import cn.hutool.core.lang.UUID;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
@Configuration
@Slf4j
public class MqttOutboundConfiguration {
@Autowired
private MqttPropertiesConfig mqttPropertiesConfig;
@Autowired
private MqttPahoClientFactory mqttPahoClientFactory;
// 消息通道
@Bean
public MessageChannel mqttOutboundChannel(){
return new DirectChannel();
}
/** 配置出站消息处理器 */
@Bean
@ServiceActivator(inputChannel = "mqttOutboundChannel") // 指定处理器针对哪个通道的消息进行处理
public MessageHandler mqttOutboundMessageHandler(){
String clientId = mqttPropertiesConfig.getPubClientId() + "_" + UUID.fastUUID();
MqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler(
mqttPropertiesConfig.getUrl(),
clientId,
mqttPahoClientFactory
);
mqttPahoMessageHandler.setDefaultQos(1);
mqttPahoMessageHandler.setDefaultTopic("B/#");
mqttPahoMessageHandler.setAsync(true);
return mqttPahoMessageHandler;
}
}

19
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/config/MqttPropertiesConfig.java
View File

@ -1,19 +0,0 @@
package com.fuyuanshen.system.mqtt.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Data
@ConfigurationProperties(prefix = "mqtt")
@Component
public class MqttPropertiesConfig {
private String username;
private String password;
private String url;
private String subClientId;
private String subTopic;
private String pubClientId;
private String pubTopic;
}

16
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/constants/MqttConstants.java
View File

@ -1,16 +0,0 @@
package com.fuyuanshen.system.mqtt.constants;
public interface MqttConstants {
/**
* 全局发布消息的key
*/
String GLOBAL_PUB_KEY = "B/";
/**
* 全局订阅消息的key
*/
String GLOBAL_SUB_KEY = "A/";
}

23
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/publish/DeviceDataController.java
View File

@ -1,23 +0,0 @@
package com.fuyuanshen.system.mqtt.publish;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/api/")
@Slf4j
public class DeviceDataController {
@Autowired
private MqttClientTest mqttClientTest;
// @PostMapping("/{deviceId}/command")
public ResponseEntity<String> sendCommand() {
mqttClientTest.sendMsg();
return ResponseEntity.ok("success");
}
}

22
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/publish/MqttClientTest.java
View File

@ -1,22 +0,0 @@
package com.fuyuanshen.system.mqtt.publish;
import com.fuyuanshen.system.mqtt.config.MqttGateway;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Slf4j
@Service
public class MqttClientTest {
@Autowired
private MqttGateway mqttGateway;
public void sendMsg() {
mqttGateway.sendMsgToMqtt("worker/location/1", "hello mqtt spring boot");
log.info("message is send");
mqttGateway.sendMsgToMqtt("worker/alert/2", "hello mqtt spring boot2");
log.info("message is send2");
}
}

25
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/publish/MqttMessageSender.java
View File

@ -1,25 +0,0 @@
package com.fuyuanshen.system.mqtt.publish;
import com.fuyuanshen.system.mqtt.config.MqttGateway;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
/**
* @Author: HarryLin
* @Date: 2025/3/20 16:16
* @Company: 北京红山信息科技研究院有限公司
* @Email: linyun@***.com.cn
**/
@Service
public class MqttMessageSender {
@Autowired
private MqttGateway mqttGateway;
public void sendMsg(@Header(value = MqttHeaders.TOPIC) String topic, String payload) {
mqttGateway.sendMsgToMqtt(topic,payload);
}
public void sendMsg(@Header(value = MqttHeaders.TOPIC) String topic, @Header(value = MqttHeaders.QOS) int qos, String payload) {
mqttGateway.sendMsgToMqtt(topic,qos,payload);
}
}

90
fys-modules/fys-system/src/main/java/com/fuyuanshen/system/mqtt/receiver/ReceiverMessageHandler.java
View File

@ -1,90 +0,0 @@
package com.fuyuanshen.system.mqtt.receiver;
import cn.hutool.core.lang.Dict;
import com.fuyuanshen.common.core.utils.ImageToCArrayConverter;
import com.fuyuanshen.common.core.utils.StringUtils;
import com.fuyuanshen.common.json.utils.JsonUtils;
import com.fuyuanshen.common.redis.utils.RedisUtils;
import com.fuyuanshen.common.satoken.utils.AppLoginHelper;
import com.fuyuanshen.common.satoken.utils.LoginHelper;
import com.fuyuanshen.system.mqtt.config.MqttGateway;
import com.fuyuanshen.system.mqtt.constants.MqttConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Service;
import java.util.*;
import static com.fuyuanshen.common.core.utils.ImageToCArrayConverter.convertHexToDecimal;
@Service
@Slf4j
public class ReceiverMessageHandler implements MessageHandler {
@Autowired
private MqttGateway mqttGateway;
@Override
public void handleMessage(Message<?> message) throws MessagingException{
String tenantId = AppLoginHelper.getTenantId() != null ? AppLoginHelper.getTenantId() : LoginHelper.getTenantId();
Object payload = message.getPayload();
MessageHeaders headers = message.getHeaders();
String receivedTopic = Objects.requireNonNull(headers.get("mqtt_receivedTopic")).toString();
String receivedQos = Objects.requireNonNull(headers.get("mqtt_receivedQos")).toString();
String timestamp = Objects.requireNonNull(headers.get("timestamp")).toString();
log.info("MQTT payload= {} \n receivedTopic = {} \n receivedQos = {} \n timestamp = {}"
,payload,receivedTopic,receivedQos,timestamp);
Dict payloadDict = JsonUtils.parseMap(payload.toString());
if(receivedTopic ==null){
return;
}
if(payloadDict == null){
return;
}
String state = payloadDict.getStr("state");
byte[] convertArr = ImageToCArrayConverter.convertStringToByteArray( state);
if(convertArr.length>0){
byte val1 = convertArr[0];
byte val2 = convertArr[1];
String[] subStr = receivedTopic.split("/");
System.out.println("收到设备id: " + subStr[1]);
String deviceImei = subStr[1];
if(val1==3){
if(val2==100){
return;
}
String data = RedisUtils.getCacheObject("894078:app_logo_data:"+deviceImei);
if(StringUtils.isEmpty(data)){
return;
}
byte[] arr = ImageToCArrayConverter.convertStringToByteArray(data);
byte[] specificChunk = ImageToCArrayConverter.getChunk(arr, (val2-1), 512);
System.out.println("第"+val2+"块数据大小: " + specificChunk.length + " 字节");
System.out.println("第"+val2+"块数据: " + Arrays.toString(specificChunk));
ArrayList<Integer> intData = new ArrayList<>();
intData.add(3);
intData.add((int) val2);
ImageToCArrayConverter.buildArr(convertHexToDecimal(specificChunk),intData);
intData.add(0);
intData.add(0);
intData.add(0);
intData.add(0);
Map<String, Object> map = new HashMap<>();
map.put("instruct", intData);
mqttGateway.sendMsgToMqtt(MqttConstants.GLOBAL_PUB_KEY+deviceImei, 1 , JsonUtils.toJsonString( map));
log.info("发送点阵数据到设备消息=>topic:{},payload:{}", MqttConstants.GLOBAL_PUB_KEY+deviceImei,JsonUtils.toJsonString(map));
}
System.out.println("val1:"+ val1);
System.out.println("val2:"+ val2);
}
}
}