Compare commits
No commits in common. '62c8a6fc87a9ece2a4b7553bc166d989889c931a' and 'ada1d2ec2fd70f153cb49074c73542f4f8068931' have entirely different histories.
62c8a6fc87
...
ada1d2ec2f
52 changed files with 30 additions and 2289 deletions
@ -1,122 +0,0 @@ |
|||||||
package com.mh.user.config; |
|
||||||
|
|
||||||
import org.springframework.boot.context.properties.ConfigurationProperties; |
|
||||||
import org.springframework.stereotype.Component; |
|
||||||
|
|
||||||
/** |
|
||||||
* 读取项目相关配置 |
|
||||||
* |
|
||||||
* @author mh |
|
||||||
*/ |
|
||||||
@Component |
|
||||||
@ConfigurationProperties(prefix = "mh") |
|
||||||
public class MHConfig |
|
||||||
{ |
|
||||||
/** 项目名称 */ |
|
||||||
private String name; |
|
||||||
|
|
||||||
/** 版本 */ |
|
||||||
private String version; |
|
||||||
|
|
||||||
/** 版权年份 */ |
|
||||||
private String copyrightYear; |
|
||||||
|
|
||||||
/** 上传路径 */ |
|
||||||
private static String profile; |
|
||||||
|
|
||||||
/** 获取地址开关 */ |
|
||||||
private static boolean addressEnabled; |
|
||||||
|
|
||||||
/** 验证码类型 */ |
|
||||||
private static String captchaType; |
|
||||||
|
|
||||||
public String getName() |
|
||||||
{ |
|
||||||
return name; |
|
||||||
} |
|
||||||
|
|
||||||
public void setName(String name) |
|
||||||
{ |
|
||||||
this.name = name; |
|
||||||
} |
|
||||||
|
|
||||||
public String getVersion() |
|
||||||
{ |
|
||||||
return version; |
|
||||||
} |
|
||||||
|
|
||||||
public void setVersion(String version) |
|
||||||
{ |
|
||||||
this.version = version; |
|
||||||
} |
|
||||||
|
|
||||||
public String getCopyrightYear() |
|
||||||
{ |
|
||||||
return copyrightYear; |
|
||||||
} |
|
||||||
|
|
||||||
public void setCopyrightYear(String copyrightYear) |
|
||||||
{ |
|
||||||
this.copyrightYear = copyrightYear; |
|
||||||
} |
|
||||||
|
|
||||||
public static String getProfile() |
|
||||||
{ |
|
||||||
return profile; |
|
||||||
} |
|
||||||
|
|
||||||
public void setProfile(String profile) |
|
||||||
{ |
|
||||||
MHConfig.profile = profile; |
|
||||||
} |
|
||||||
|
|
||||||
public static boolean isAddressEnabled() |
|
||||||
{ |
|
||||||
return addressEnabled; |
|
||||||
} |
|
||||||
|
|
||||||
public void setAddressEnabled(boolean addressEnabled) |
|
||||||
{ |
|
||||||
MHConfig.addressEnabled = addressEnabled; |
|
||||||
} |
|
||||||
|
|
||||||
public static String getCaptchaType() { |
|
||||||
return captchaType; |
|
||||||
} |
|
||||||
|
|
||||||
public void setCaptchaType(String captchaType) { |
|
||||||
MHConfig.captchaType = captchaType; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取导入上传路径 |
|
||||||
*/ |
|
||||||
public static String getImportPath() |
|
||||||
{ |
|
||||||
return getProfile() + "/import"; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取头像上传路径 |
|
||||||
*/ |
|
||||||
public static String getAvatarPath() |
|
||||||
{ |
|
||||||
return getProfile() + "/avatar"; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取下载路径 |
|
||||||
*/ |
|
||||||
public static String getDownloadPath() |
|
||||||
{ |
|
||||||
return getProfile() + "/download/"; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取上传路径 |
|
||||||
*/ |
|
||||||
public static String getUploadPath() |
|
||||||
{ |
|
||||||
return getProfile() + "/upload"; |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,62 +0,0 @@ |
|||||||
package com.mh.user.config.mqtt; |
|
||||||
|
|
||||||
import com.mh.user.config.MHConfig; |
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import com.mh.user.constants.TopicEnum; |
|
||||||
import com.mh.user.utils.SpringBeanUtil; |
|
||||||
import lombok.extern.slf4j.Slf4j; |
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||||
import org.springframework.integration.annotation.Router; |
|
||||||
import org.springframework.integration.mqtt.support.MqttHeaders; |
|
||||||
import org.springframework.integration.router.AbstractMessageRouter; |
|
||||||
import org.springframework.messaging.Message; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
import org.springframework.messaging.MessageHeaders; |
|
||||||
import org.springframework.stereotype.Component; |
|
||||||
|
|
||||||
import java.util.Collection; |
|
||||||
import java.util.Collections; |
|
||||||
import java.util.Objects; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 入站消息路由分发中心 |
|
||||||
* @date 2024-10-29 17:04:17 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Component |
|
||||||
public class InboundMessageRouter extends AbstractMessageRouter { |
|
||||||
|
|
||||||
/** 系统基础配置 */ |
|
||||||
@Autowired |
|
||||||
private MHConfig mHConfig; |
|
||||||
|
|
||||||
/** |
|
||||||
* 目前只需要这个方式，后期在拓展使用IntegrationFlow方式 |
|
||||||
* @param message |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Router(inputChannel = ChannelName.INBOUND) |
|
||||||
@Override |
|
||||||
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) { |
|
||||||
MessageHeaders headers = message.getHeaders(); |
|
||||||
String topic = Objects.requireNonNull(headers.get(MqttHeaders.RECEIVED_TOPIC)).toString(); |
|
||||||
// byte[] payload = (byte[]) message.getPayload();
|
|
||||||
// log.info("从当前主题 topic: {}, 接收到的消息：{}", topic, new String(payload));
|
|
||||||
// 判断当前主题是否是当前项目的,温湿度目前写死的
|
|
||||||
if (!topic.startsWith(mHConfig.getName()) && !topic.contains("/temp")) { |
|
||||||
log.info("当前主题 topic: {} 不是当前项目的，直接丢弃", topic); |
|
||||||
return Collections.singleton((MessageChannel) SpringBeanUtil.getBean(ChannelName.DEFAULT_BOUND)); |
|
||||||
} |
|
||||||
// 找到对应的主题消息通道
|
|
||||||
if (topic.contains("/temp")) { |
|
||||||
return Collections.singleton((MessageChannel) SpringBeanUtil.getBean(ChannelName.EVENTS_UPLOAD_INBOUND)); |
|
||||||
} else { |
|
||||||
TopicEnum topicEnum = TopicEnum.find(mHConfig.getName() + "/", topic); |
|
||||||
MessageChannel bean = (MessageChannel) SpringBeanUtil.getBean(topicEnum.getBeanName()); |
|
||||||
return Collections.singleton(bean); |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,99 +0,0 @@ |
|||||||
package com.mh.user.config.mqtt; |
|
||||||
|
|
||||||
import com.mh.user.constants.MqttClientOptions; |
|
||||||
import com.mh.user.constants.MqttProtocolEnum; |
|
||||||
import com.mh.user.constants.MqttUseEnum; |
|
||||||
import lombok.Data; |
|
||||||
import org.eclipse.paho.client.mqttv3.MqttConnectOptions; |
|
||||||
import org.springframework.boot.context.properties.ConfigurationProperties; |
|
||||||
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; |
|
||||||
import org.springframework.util.StringUtils; |
|
||||||
|
|
||||||
import java.util.Map; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description mqtt连接配置 |
|
||||||
* @date 2024-10-29 14:44:51 |
|
||||||
*/ |
|
||||||
@Configuration |
|
||||||
@Data |
|
||||||
@ConfigurationProperties |
|
||||||
public class MqttConfig { |
|
||||||
|
|
||||||
private static Map<MqttUseEnum, MqttClientOptions> mqttSpring; |
|
||||||
|
|
||||||
public void setMqttSpring(Map<MqttUseEnum, MqttClientOptions> mqtt) { |
|
||||||
MqttConfig.mqttSpring = mqtt; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取mqtt基本配置 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
static MqttClientOptions getBasicMqttClientOptions() { |
|
||||||
if (!mqttSpring.containsKey(MqttUseEnum.BASIC)) { |
|
||||||
throw new Error("请先配置MQTT的基本连接参数，否则无法启动项目"); |
|
||||||
} |
|
||||||
return mqttSpring.get(MqttUseEnum.BASIC); |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 拼接获取对应mqtt的连接地址 |
|
||||||
* @param options |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
public static String getMqttAddress(MqttClientOptions options) { |
|
||||||
StringBuilder addr = new StringBuilder(); |
|
||||||
addr.append(options.getProtocol().getProtocolAddr()) |
|
||||||
.append(options.getHost().trim()) |
|
||||||
.append(":") |
|
||||||
.append(options.getPort()); |
|
||||||
if ((options.getProtocol() == MqttProtocolEnum.WS || options.getProtocol() == MqttProtocolEnum.WSS) |
|
||||||
&& StringUtils.hasText(options.getPath())) { |
|
||||||
addr.append(options.getPath()); |
|
||||||
} |
|
||||||
return addr.toString(); |
|
||||||
} |
|
||||||
|
|
||||||
public static String getBasicMqttAddress() { |
|
||||||
return getMqttAddress(getBasicMqttClientOptions()); |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取连接参数，注入到spring中 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean |
|
||||||
public MqttConnectOptions mqttConnectionOptions() { |
|
||||||
|
|
||||||
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions(); |
|
||||||
|
|
||||||
MqttClientOptions customizeOptions = getBasicMqttClientOptions(); |
|
||||||
String basicMqttAddress = getBasicMqttAddress(); |
|
||||||
mqttConnectOptions.setServerURIs(new String[]{basicMqttAddress}); |
|
||||||
mqttConnectOptions.setUserName(StringUtils.hasText(customizeOptions.getUsername()) ? |
|
||||||
customizeOptions.getUsername() : ""); |
|
||||||
mqttConnectOptions.setPassword(StringUtils.hasText(customizeOptions.getPassword()) ? |
|
||||||
customizeOptions.getPassword().toCharArray() : new char[0]); |
|
||||||
// 直接进行自动连接
|
|
||||||
mqttConnectOptions.setAutomaticReconnect(true); |
|
||||||
// 时间间隔时间10s
|
|
||||||
mqttConnectOptions.setKeepAliveInterval(10); |
|
||||||
|
|
||||||
return mqttConnectOptions; |
|
||||||
} |
|
||||||
|
|
||||||
@Bean |
|
||||||
public MqttPahoClientFactory mqttClientFactory() { |
|
||||||
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); |
|
||||||
factory.setConnectionOptions(mqttConnectionOptions()); |
|
||||||
return factory; |
|
||||||
} |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,91 +0,0 @@ |
|||||||
package com.mh.user.config.mqtt; |
|
||||||
|
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import com.mh.user.constants.MqttClientOptions; |
|
||||||
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.IntegrationComponentScan; |
|
||||||
import org.springframework.integration.annotation.ServiceActivator; |
|
||||||
import org.springframework.integration.endpoint.MessageProducerSupport; |
|
||||||
import org.springframework.integration.mqtt.core.MqttPahoClientFactory; |
|
||||||
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; |
|
||||||
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; |
|
||||||
import org.springframework.integration.mqtt.support.MqttHeaders; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
import org.springframework.messaging.MessageHandler; |
|
||||||
|
|
||||||
import javax.annotation.Resource; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 入站配置 |
|
||||||
* @date 2024-10-29 16:22:10 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Configuration |
|
||||||
@IntegrationComponentScan |
|
||||||
public class MqttInboundConfig { |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private MqttPahoClientFactory mqttClientFactory; |
|
||||||
|
|
||||||
@Resource(name = ChannelName.INBOUND) |
|
||||||
private MessageChannel inboundChannel; |
|
||||||
|
|
||||||
private String clientId; |
|
||||||
|
|
||||||
/** |
|
||||||
* 入站适配器 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean(name = "adapter") |
|
||||||
public MessageProducerSupport mqttInbound() { |
|
||||||
MqttClientOptions options = MqttConfig.getBasicMqttClientOptions(); |
|
||||||
// 此处初始化的时候，默认订阅了配置文件中已经写好的topic
|
|
||||||
// 如果需要订阅多个，可以自己手动订阅，会写一个addTopic()进行添加订阅
|
|
||||||
clientId = options.getClientId() + "_consumer_" + System.currentTimeMillis(); |
|
||||||
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter( |
|
||||||
clientId, |
|
||||||
mqttClientFactory, |
|
||||||
options.getInboundTopic().split(",")); |
|
||||||
// System.out.println("每一次都会入站适配器吗？"+clientId);
|
|
||||||
DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(); |
|
||||||
// 统一是字节处理
|
|
||||||
converter.setPayloadAsBytes(true); |
|
||||||
// 设置消息转换器
|
|
||||||
adapter.setConverter(converter); |
|
||||||
// 设置qos（quality of service）
|
|
||||||
// 0：最多一次传输（消息会丢失），
|
|
||||||
// 1：至少一次传输（消息会重复），
|
|
||||||
// 2：只有当消息发送成功时才确认（消息不回丢，但延迟高）。
|
|
||||||
adapter.setQos(0); |
|
||||||
// 设置在接收已经订阅的主题信息后，发送给哪个通道，具体的发送方法需要翻上层的抽象类
|
|
||||||
adapter.setOutputChannel(inboundChannel); |
|
||||||
return adapter; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 默认声明一个消息处理器，用于处理无效的消息 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean |
|
||||||
@ServiceActivator(inputChannel = ChannelName.DEFAULT_BOUND) |
|
||||||
public MessageHandler handler() { |
|
||||||
return message -> { |
|
||||||
log.info("The default channel does not handle messages." + |
|
||||||
"\nTopic: {}" + |
|
||||||
"\nPayload: {}", |
|
||||||
message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC), |
|
||||||
message.getPayload()); |
|
||||||
}; |
|
||||||
} |
|
||||||
|
|
||||||
public String getClientId() { |
|
||||||
return clientId; |
|
||||||
} |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,55 +0,0 @@ |
|||||||
package com.mh.user.config.mqtt; |
|
||||||
|
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import lombok.extern.slf4j.Slf4j; |
|
||||||
import org.springframework.context.annotation.Bean; |
|
||||||
import org.springframework.context.annotation.Configuration; |
|
||||||
import org.springframework.integration.channel.DirectChannel; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 声明所有通道的定义类 |
|
||||||
* @date 2024-10-29 16:23:32 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Configuration |
|
||||||
public class MqttMessageChannel { |
|
||||||
|
|
||||||
@Bean(name = ChannelName.OUTBOUND) |
|
||||||
public MessageChannel outboundChannel() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.INBOUND) |
|
||||||
public MessageChannel inboundChannel() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 事件主动上报通道 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean(name = ChannelName.EVENTS_UPLOAD_INBOUND) |
|
||||||
public MessageChannel eventsUploadInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.EVENTS_COLLECTION_INBOUND) |
|
||||||
public MessageChannel eventsCollectionInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.EVENTS_CONTROL_INBOUND) |
|
||||||
public MessageChannel eventsControlInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.EVENTS_SEND_INBOUND) |
|
||||||
public MessageChannel eventsSendInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,51 +0,0 @@ |
|||||||
package com.mh.user.config.mqtt; |
|
||||||
|
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
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.IntegrationComponentScan; |
|
||||||
import org.springframework.integration.annotation.ServiceActivator; |
|
||||||
import org.springframework.integration.mqtt.core.MqttPahoClientFactory; |
|
||||||
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; |
|
||||||
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; |
|
||||||
import org.springframework.messaging.MessageHandler; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 入站配置 |
|
||||||
* @date 2024-10-29 16:22:10 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Configuration |
|
||||||
@IntegrationComponentScan |
|
||||||
public class MqttOutboundConfig { |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private MqttPahoClientFactory mqttClientFactory; |
|
||||||
|
|
||||||
/** |
|
||||||
* 默认声明一个出站处理器，用于处理无效的消息 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean |
|
||||||
@ServiceActivator(inputChannel = ChannelName.OUTBOUND) |
|
||||||
public MessageHandler mqttOutbound() { |
|
||||||
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler( |
|
||||||
MqttConfig.getBasicMqttClientOptions().getClientId() + "_producer_" + System.currentTimeMillis(), |
|
||||||
mqttClientFactory); |
|
||||||
DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(); |
|
||||||
// use byte types uniformly
|
|
||||||
converter.setPayloadAsBytes(true); |
|
||||||
|
|
||||||
messageHandler.setAsync(true); |
|
||||||
messageHandler.setDefaultQos(0); |
|
||||||
messageHandler.setConverter(converter); |
|
||||||
return messageHandler; |
|
||||||
} |
|
||||||
|
|
||||||
|
|
||||||
} |
|
||||||
@ -1,59 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 声明所有通道 |
|
||||||
* @date 2024-10-29 16:04:19 |
|
||||||
*/ |
|
||||||
public class ChannelName { |
|
||||||
|
|
||||||
/** |
|
||||||
* 默认通道名称（防止出错） |
|
||||||
*/ |
|
||||||
public static final String DEFAULT_BOUND = "default_bound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 主动上报入站 |
|
||||||
*/ |
|
||||||
public static final String INBOUND = "inbound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 出站 |
|
||||||
*/ |
|
||||||
public static final String OUTBOUND = "outbound"; |
|
||||||
|
|
||||||
|
|
||||||
/** |
|
||||||
* 入站主动上报 |
|
||||||
*/ |
|
||||||
public static final String EVENTS_UPLOAD_INBOUND = "events_upload_inbound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 入站主动采集 |
|
||||||
*/ |
|
||||||
public static final String EVENTS_COLLECTION_INBOUND = "events_collection_inbound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 入站主动控制 |
|
||||||
*/ |
|
||||||
public static final String EVENTS_CONTROL_INBOUND = "events_control_inbound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 默认进站处理 |
|
||||||
*/ |
|
||||||
public static final String EVENTS_DEFAULT_INBOUND = "events_default_inbound"; |
|
||||||
|
|
||||||
public static final String REPLY_EVENTS_OUTBOUND = "reply_events_outbound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 新珠江收到的信息 |
|
||||||
*/ |
|
||||||
public static final String EVENTS_RECEIVE_INBOUND = "events_receive_inbound"; |
|
||||||
|
|
||||||
/** |
|
||||||
* 接收服务端的数据报文 |
|
||||||
*/ |
|
||||||
public static final String EVENTS_SEND_INBOUND = "events_send_inbound"; |
|
||||||
} |
|
||||||
@ -1,36 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|
||||||
import lombok.AllArgsConstructor; |
|
||||||
import lombok.Builder; |
|
||||||
import lombok.Data; |
|
||||||
import lombok.NoArgsConstructor; |
|
||||||
|
|
||||||
/** |
|
||||||
* Unified Topic response format |
|
||||||
* |
|
||||||
* @author sean.zhou |
|
||||||
* @version 0.1 |
|
||||||
* @date 2021/11/15 |
|
||||||
*/ |
|
||||||
@Data |
|
||||||
@AllArgsConstructor |
|
||||||
@NoArgsConstructor |
|
||||||
@Builder |
|
||||||
@JsonIgnoreProperties(ignoreUnknown = true) |
|
||||||
public class CommonTopicResponse<T> { |
|
||||||
|
|
||||||
/** |
|
||||||
* The command is sent and the response is matched by the tid and bid fields in the message, |
|
||||||
* and the reply should keep the tid and bid the same. |
|
||||||
*/ |
|
||||||
private String tid; |
|
||||||
|
|
||||||
private String bid; |
|
||||||
|
|
||||||
private String method; |
|
||||||
|
|
||||||
private T data; |
|
||||||
|
|
||||||
private Long timestamp; |
|
||||||
} |
|
||||||
@ -1,108 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description mqtt连接的参数 |
|
||||||
* @date 2024-10-29 14:46:24 |
|
||||||
*/ |
|
||||||
public class MqttClientOptions { |
|
||||||
|
|
||||||
private MqttProtocolEnum protocol; |
|
||||||
|
|
||||||
private String host; |
|
||||||
|
|
||||||
private Integer port; |
|
||||||
|
|
||||||
private String username; |
|
||||||
|
|
||||||
private String password; |
|
||||||
|
|
||||||
private String clientId; |
|
||||||
|
|
||||||
private String path; |
|
||||||
|
|
||||||
/** |
|
||||||
* 客户端连接的时候，订阅的主题 |
|
||||||
*/ |
|
||||||
private String inboundTopic; |
|
||||||
|
|
||||||
public MqttProtocolEnum getProtocol() { |
|
||||||
return protocol; |
|
||||||
} |
|
||||||
|
|
||||||
public void setProtocol(MqttProtocolEnum protocol) { |
|
||||||
this.protocol = protocol; |
|
||||||
} |
|
||||||
|
|
||||||
public String getHost() { |
|
||||||
return host; |
|
||||||
} |
|
||||||
|
|
||||||
public void setHost(String host) { |
|
||||||
this.host = host; |
|
||||||
} |
|
||||||
|
|
||||||
public Integer getPort() { |
|
||||||
return port; |
|
||||||
} |
|
||||||
|
|
||||||
public void setPort(Integer port) { |
|
||||||
this.port = port; |
|
||||||
} |
|
||||||
|
|
||||||
public String getUsername() { |
|
||||||
return username; |
|
||||||
} |
|
||||||
|
|
||||||
public void setUsername(String username) { |
|
||||||
this.username = username; |
|
||||||
} |
|
||||||
|
|
||||||
public String getPassword() { |
|
||||||
return password; |
|
||||||
} |
|
||||||
|
|
||||||
public void setPassword(String password) { |
|
||||||
this.password = password; |
|
||||||
} |
|
||||||
|
|
||||||
public String getClientId() { |
|
||||||
return clientId; |
|
||||||
} |
|
||||||
|
|
||||||
public void setClientId(String clientId) { |
|
||||||
this.clientId = clientId; |
|
||||||
} |
|
||||||
|
|
||||||
public String getPath() { |
|
||||||
return path; |
|
||||||
} |
|
||||||
|
|
||||||
public void setPath(String path) { |
|
||||||
this.path = path; |
|
||||||
} |
|
||||||
|
|
||||||
public String getInboundTopic() { |
|
||||||
return inboundTopic; |
|
||||||
} |
|
||||||
|
|
||||||
public void setInboundTopic(String inboundTopic) { |
|
||||||
this.inboundTopic = inboundTopic; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public String toString() { |
|
||||||
return "MqttClientOptions{" + |
|
||||||
"protocol=" + protocol + |
|
||||||
", host='" + host + '\'' + |
|
||||||
", port=" + port + |
|
||||||
", username='" + username + '\'' + |
|
||||||
", password='" + password + '\'' + |
|
||||||
", clientId='" + clientId + '\'' + |
|
||||||
", path='" + path + '\'' + |
|
||||||
", inboundTopic='" + inboundTopic + '\'' + |
|
||||||
'}'; |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,33 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 采用哪种协议进行数据交互 |
|
||||||
* @date 2024-10-29 15:21:07 |
|
||||||
*/ |
|
||||||
public enum MqttProtocolEnum { |
|
||||||
|
|
||||||
MQTT("tcp"), |
|
||||||
|
|
||||||
MQTTS("tcp"), |
|
||||||
|
|
||||||
WS("ws"), |
|
||||||
|
|
||||||
WSS("wss"); |
|
||||||
|
|
||||||
final String protocol; |
|
||||||
|
|
||||||
MqttProtocolEnum(String protocol) { |
|
||||||
this.protocol = protocol; |
|
||||||
} |
|
||||||
|
|
||||||
public String getProtocolAddr() { |
|
||||||
return protocol + "://"; |
|
||||||
} |
|
||||||
|
|
||||||
public String getProtocol() { |
|
||||||
return protocol; |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,16 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description mqtt选择模式 |
|
||||||
* @date 2024-10-29 15:19:21 |
|
||||||
*/ |
|
||||||
public enum MqttUseEnum { |
|
||||||
|
|
||||||
BASIC, |
|
||||||
|
|
||||||
DRC |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,20 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|
||||||
import lombok.Data; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author sean.zhou |
|
||||||
* @version 0.1 |
|
||||||
* @date 2021/11/22 |
|
||||||
*/ |
|
||||||
@Data |
|
||||||
@JsonIgnoreProperties(ignoreUnknown = true) |
|
||||||
public class ServiceReply<T> { |
|
||||||
|
|
||||||
private Integer result; |
|
||||||
|
|
||||||
private T info; |
|
||||||
|
|
||||||
private T output; |
|
||||||
} |
|
||||||
@ -1,31 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
/** |
|
||||||
* All the topics that need to be used in the project. |
|
||||||
* |
|
||||||
* @author ljf |
|
||||||
* @version 0.1 |
|
||||||
* @date 2025-01-22 |
|
||||||
*/ |
|
||||||
public class TopicConst { |
|
||||||
|
|
||||||
public static final String MH_UPLOAD = "mh_upload/"; |
|
||||||
|
|
||||||
public static final String EVENTS_UPLOAD = "events_upload/"; |
|
||||||
|
|
||||||
public static final String MH_COLLECTION = "mh_collection/"; |
|
||||||
|
|
||||||
public static final String EVENTS_COLLECTION = "events_collection/"; |
|
||||||
|
|
||||||
public static final String MH_CONTROL = "mh_control/"; |
|
||||||
|
|
||||||
public static final String EVENTS_CONTROL = "events_control/"; |
|
||||||
|
|
||||||
public static final String REGEX_SN = "[A-Za-z0-9]+"; |
|
||||||
|
|
||||||
public static final String THING_MODEL_PRE = "thing/"; |
|
||||||
|
|
||||||
public static final String PRODUCT = "product/"; |
|
||||||
|
|
||||||
public static final String SERVICES_SUF = "/services"; |
|
||||||
} |
|
||||||
@ -1,60 +0,0 @@ |
|||||||
package com.mh.user.constants; |
|
||||||
|
|
||||||
import java.util.Arrays; |
|
||||||
import java.util.regex.Pattern; |
|
||||||
|
|
||||||
import static com.mh.user.constants.TopicConst.*; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author ljf |
|
||||||
* @version 1.0 |
|
||||||
* @description: TODO |
|
||||||
* @date 2024/11/06 14:28 |
|
||||||
*/ |
|
||||||
public enum TopicEnum { |
|
||||||
|
|
||||||
/** |
|
||||||
* 冷水机组客户端主动上报数据 |
|
||||||
*/ |
|
||||||
CLIENT_UPLOAD_DATA(Pattern.compile("^" + MH_UPLOAD + EVENTS_UPLOAD + REGEX_SN + "$"), ChannelName.EVENTS_UPLOAD_INBOUND), |
|
||||||
|
|
||||||
/** |
|
||||||
* 服务端采集数据 |
|
||||||
*/ |
|
||||||
SERVER_COLLECTION_DATA(Pattern.compile("^" + MH_COLLECTION + EVENTS_COLLECTION + REGEX_SN + "$"), ChannelName.EVENTS_COLLECTION_INBOUND), |
|
||||||
|
|
||||||
/** |
|
||||||
* 服务端控制指令 |
|
||||||
*/ |
|
||||||
SERVER_CONTROL_DATA(Pattern.compile("^" + MH_CONTROL + EVENTS_CONTROL + REGEX_SN + "$"), ChannelName.EVENTS_CONTROL_INBOUND), |
|
||||||
|
|
||||||
/** |
|
||||||
* 订阅服务端发送的主题命令 |
|
||||||
*/ |
|
||||||
SERVER_SEND_DATA(Pattern.compile("^A/cmd/ctl/send" + "$"), ChannelName.EVENTS_SEND_INBOUND), |
|
||||||
|
|
||||||
UNKNOWN(Pattern.compile("^.*$"), ChannelName.DEFAULT_BOUND); |
|
||||||
|
|
||||||
final Pattern pattern; |
|
||||||
|
|
||||||
final String beanName; |
|
||||||
|
|
||||||
TopicEnum(Pattern pattern, String beanName) { |
|
||||||
this.pattern = pattern; |
|
||||||
this.beanName = beanName; |
|
||||||
} |
|
||||||
|
|
||||||
public Pattern getPattern() { |
|
||||||
return pattern; |
|
||||||
} |
|
||||||
|
|
||||||
public String getBeanName() { |
|
||||||
return beanName; |
|
||||||
} |
|
||||||
|
|
||||||
public static TopicEnum find(String proName, String topic) { |
|
||||||
// 去掉第一个"/"以及之前数据
|
|
||||||
String finalTopic = topic.replaceFirst("^"+proName, "");; |
|
||||||
return Arrays.stream(TopicEnum.values()).filter(topicEnum -> topicEnum.pattern.matcher(finalTopic).matches()).findAny().orElse(UNKNOWN); |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,48 +0,0 @@ |
|||||||
package com.mh.user.entity; |
|
||||||
|
|
||||||
import com.fasterxml.jackson.annotation.JsonFormat; |
|
||||||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|
||||||
import com.fasterxml.jackson.annotation.JsonInclude; |
|
||||||
import lombok.Data; |
|
||||||
|
|
||||||
import java.util.Date; |
|
||||||
import java.util.Map; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description mqtt订阅管理 |
|
||||||
* @date 2025-02-14 13:47:07 |
|
||||||
*/ |
|
||||||
@Data |
|
||||||
public class MqttSubscriptionEntity { |
|
||||||
|
|
||||||
private Long id; |
|
||||||
|
|
||||||
private String topic; |
|
||||||
|
|
||||||
private Short qos; |
|
||||||
|
|
||||||
private String clientId; |
|
||||||
|
|
||||||
private String status; |
|
||||||
|
|
||||||
/** 创建者 */ |
|
||||||
private String createBy; |
|
||||||
|
|
||||||
/** 创建时间 */ |
|
||||||
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") |
|
||||||
private Date createTime; |
|
||||||
|
|
||||||
/** 更新者 */ |
|
||||||
private String updateBy; |
|
||||||
|
|
||||||
/** 更新时间 */ |
|
||||||
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") |
|
||||||
private Date updateTime; |
|
||||||
|
|
||||||
/** 备注 */ |
|
||||||
private String remark; |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,32 +0,0 @@ |
|||||||
package com.mh.user.mapper; |
|
||||||
|
|
||||||
import com.mh.user.entity.MqttSubscriptionEntity; |
|
||||||
import org.apache.ibatis.annotations.Mapper; |
|
||||||
import org.apache.ibatis.annotations.Param; |
|
||||||
import org.apache.ibatis.annotations.Select; |
|
||||||
import tk.mybatis.mapper.common.BaseMapper; |
|
||||||
|
|
||||||
import java.util.List; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description mqtt订阅mapper类 |
|
||||||
* @date 2025-02-14 14:00:58 |
|
||||||
*/ |
|
||||||
@Mapper |
|
||||||
public interface MqttSubscriptionMapper extends BaseMapper<MqttSubscriptionEntity> { |
|
||||||
|
|
||||||
@Select("<script>" + |
|
||||||
"SELECT * FROM mqtt_subscription WHERE 1=1 " + |
|
||||||
"<if test='topic != null and topic != \"\"'>AND topic = #{topic}</if>" + |
|
||||||
"<if test='status != null and status != \"\"'>AND status = #{status}</if>" + |
|
||||||
"ORDER BY crate_time DESC " + |
|
||||||
"</script>") |
|
||||||
List<MqttSubscriptionEntity> selectListByTopic(@Param("topic") String topic, |
|
||||||
@Param("status") String status); |
|
||||||
|
|
||||||
@Select("SELECT top 1 * FROM mqtt_subscription WHERE id = #{id}") |
|
||||||
MqttSubscriptionEntity selectById(@Param("id") String id); |
|
||||||
} |
|
||||||
@ -1,30 +0,0 @@ |
|||||||
package com.mh.user.model; |
|
||||||
|
|
||||||
import lombok.Data; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description 研华数据体 |
|
||||||
* @date 2025-01-22 14:47:25 |
|
||||||
*/ |
|
||||||
@Data |
|
||||||
public class SanShiFengDatas<T extends Number> { |
|
||||||
|
|
||||||
/** |
|
||||||
* 对应研华的标签值 |
|
||||||
*/ |
|
||||||
private String tag; |
|
||||||
|
|
||||||
/** |
|
||||||
* 上报值 |
|
||||||
*/ |
|
||||||
private T value; |
|
||||||
|
|
||||||
/** |
|
||||||
* 质量值 |
|
||||||
*/ |
|
||||||
private T quality; |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,29 +0,0 @@ |
|||||||
package com.mh.user.model; |
|
||||||
|
|
||||||
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; |
|
||||||
import lombok.Data; |
|
||||||
|
|
||||||
import java.util.List; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description 研华网关发送接收数据 |
|
||||||
* @date 2025-01-22 14:43:15 |
|
||||||
*/ |
|
||||||
@Data |
|
||||||
@JsonIgnoreProperties(ignoreUnknown = true) |
|
||||||
public class SanShiFengReceiver<T> { |
|
||||||
|
|
||||||
/** |
|
||||||
* 数据集合 |
|
||||||
*/ |
|
||||||
private List<T> d; |
|
||||||
|
|
||||||
/** |
|
||||||
* 主动上报数据时间（带T类型） |
|
||||||
*/ |
|
||||||
private String ts; |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,26 +0,0 @@ |
|||||||
package com.mh.user.service; |
|
||||||
|
|
||||||
import com.mh.user.entity.MqttSubscriptionEntity; |
|
||||||
|
|
||||||
import java.util.List; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description mqtt订阅管理 |
|
||||||
* @date 2025-02-14 13:58:37 |
|
||||||
*/ |
|
||||||
public interface MqttSubscriptionService { |
|
||||||
|
|
||||||
List<MqttSubscriptionEntity> selectMqttSubList(MqttSubscriptionEntity mqttSubscription); |
|
||||||
|
|
||||||
MqttSubscriptionEntity selectMqttSubById(String msId); |
|
||||||
|
|
||||||
int insertMqttSub(MqttSubscriptionEntity mqttSubscription); |
|
||||||
|
|
||||||
int updateMqttSub(MqttSubscriptionEntity mqttSubscription); |
|
||||||
|
|
||||||
int deleteMqttSubByIds(String[] msIds); |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,61 +0,0 @@ |
|||||||
package com.mh.user.service.impl; |
|
||||||
|
|
||||||
import com.mh.user.entity.MqttSubscriptionEntity; |
|
||||||
import com.mh.user.mapper.MqttSubscriptionMapper; |
|
||||||
import com.mh.user.service.MqttSubscriptionService; |
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||||
import org.springframework.stereotype.Service; |
|
||||||
|
|
||||||
import java.util.List; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description mqtt订阅实现类 |
|
||||||
* @date 2025-02-14 13:59:27 |
|
||||||
*/ |
|
||||||
@Service |
|
||||||
public class MqttSubscriptionServiceImpl implements MqttSubscriptionService { |
|
||||||
|
|
||||||
private final MqttSubscriptionMapper mqttSubscriptionMapper; |
|
||||||
|
|
||||||
@Autowired |
|
||||||
public MqttSubscriptionServiceImpl(MqttSubscriptionMapper mqttSubscriptionMapper) { |
|
||||||
this.mqttSubscriptionMapper = mqttSubscriptionMapper; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public List<MqttSubscriptionEntity> selectMqttSubList(MqttSubscriptionEntity mqttSubscription) { |
|
||||||
if (mqttSubscription == null) { |
|
||||||
return null; |
|
||||||
} |
|
||||||
return mqttSubscriptionMapper.selectListByTopic(mqttSubscription.getTopic(), mqttSubscription.getStatus()); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public MqttSubscriptionEntity selectMqttSubById(String msId) { |
|
||||||
return mqttSubscriptionMapper.selectById(msId); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public int insertMqttSub(MqttSubscriptionEntity mqttSubscription) { |
|
||||||
return mqttSubscriptionMapper.insert(mqttSubscription); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public int updateMqttSub(MqttSubscriptionEntity mqttSubscription) { |
|
||||||
return mqttSubscriptionMapper.updateByPrimaryKey(mqttSubscription); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public int deleteMqttSubByIds(String[] msIds) { |
|
||||||
if (msIds != null && msIds.length > 0) { |
|
||||||
for (String msId : msIds) { |
|
||||||
mqttSubscriptionMapper.deleteByPrimaryKey(msId); |
|
||||||
} |
|
||||||
return msIds.length; |
|
||||||
} |
|
||||||
return 0; |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,99 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.config; |
|
||||||
|
|
||||||
import com.mh.user.constants.MqttClientOptions; |
|
||||||
import com.mh.user.constants.MqttProtocolEnum; |
|
||||||
import com.mh.user.constants.MqttUseEnum; |
|
||||||
import lombok.Data; |
|
||||||
import org.eclipse.paho.client.mqttv3.MqttConnectOptions; |
|
||||||
import org.springframework.boot.context.properties.ConfigurationProperties; |
|
||||||
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; |
|
||||||
import org.springframework.util.StringUtils; |
|
||||||
|
|
||||||
import java.util.Map; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description mqtt连接配置 |
|
||||||
* @date 2024-10-29 14:44:51 |
|
||||||
*/ |
|
||||||
@Configuration |
|
||||||
@Data |
|
||||||
@ConfigurationProperties |
|
||||||
public class MqttConfig { |
|
||||||
|
|
||||||
private static Map<MqttUseEnum, MqttClientOptions> mqttSpring; |
|
||||||
|
|
||||||
public void setMqttSpring(Map<MqttUseEnum, MqttClientOptions> mqtt) { |
|
||||||
MqttConfig.mqttSpring = mqtt; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取mqtt基本配置 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
static MqttClientOptions getBasicMqttClientOptions() { |
|
||||||
if (!mqttSpring.containsKey(MqttUseEnum.BASIC)) { |
|
||||||
throw new Error("请先配置MQTT的基本连接参数，否则无法启动项目"); |
|
||||||
} |
|
||||||
return mqttSpring.get(MqttUseEnum.BASIC); |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 拼接获取对应mqtt的连接地址 |
|
||||||
* @param options |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
public static String getMqttAddress(MqttClientOptions options) { |
|
||||||
StringBuilder addr = new StringBuilder(); |
|
||||||
addr.append(options.getProtocol().getProtocolAddr()) |
|
||||||
.append(options.getHost().trim()) |
|
||||||
.append(":") |
|
||||||
.append(options.getPort()); |
|
||||||
if ((options.getProtocol() == MqttProtocolEnum.WS || options.getProtocol() == MqttProtocolEnum.WSS) |
|
||||||
&& StringUtils.hasText(options.getPath())) { |
|
||||||
addr.append(options.getPath()); |
|
||||||
} |
|
||||||
return addr.toString(); |
|
||||||
} |
|
||||||
|
|
||||||
public static String getBasicMqttAddress() { |
|
||||||
return getMqttAddress(getBasicMqttClientOptions()); |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取连接参数，注入到spring中 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean |
|
||||||
public MqttConnectOptions mqttConnectionOptions() { |
|
||||||
|
|
||||||
MqttConnectOptions mqttConnectOptions = new MqttConnectOptions(); |
|
||||||
|
|
||||||
MqttClientOptions customizeOptions = getBasicMqttClientOptions(); |
|
||||||
String basicMqttAddress = getBasicMqttAddress(); |
|
||||||
mqttConnectOptions.setServerURIs(new String[]{basicMqttAddress}); |
|
||||||
mqttConnectOptions.setUserName(StringUtils.hasText(customizeOptions.getUsername()) ? |
|
||||||
customizeOptions.getUsername() : ""); |
|
||||||
mqttConnectOptions.setPassword(StringUtils.hasText(customizeOptions.getPassword()) ? |
|
||||||
customizeOptions.getPassword().toCharArray() : new char[0]); |
|
||||||
// 直接进行自动连接
|
|
||||||
mqttConnectOptions.setAutomaticReconnect(true); |
|
||||||
// 时间间隔时间10s
|
|
||||||
mqttConnectOptions.setKeepAliveInterval(10); |
|
||||||
|
|
||||||
return mqttConnectOptions; |
|
||||||
} |
|
||||||
|
|
||||||
@Bean |
|
||||||
public MqttPahoClientFactory mqttClientFactory() { |
|
||||||
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); |
|
||||||
factory.setConnectionOptions(mqttConnectionOptions()); |
|
||||||
return factory; |
|
||||||
} |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,91 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.config; |
|
||||||
|
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import com.mh.user.constants.MqttClientOptions; |
|
||||||
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.IntegrationComponentScan; |
|
||||||
import org.springframework.integration.annotation.ServiceActivator; |
|
||||||
import org.springframework.integration.endpoint.MessageProducerSupport; |
|
||||||
import org.springframework.integration.mqtt.core.MqttPahoClientFactory; |
|
||||||
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; |
|
||||||
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; |
|
||||||
import org.springframework.integration.mqtt.support.MqttHeaders; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
import org.springframework.messaging.MessageHandler; |
|
||||||
|
|
||||||
import javax.annotation.Resource; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 入站配置 |
|
||||||
* @date 2024-10-29 16:22:10 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Configuration |
|
||||||
@IntegrationComponentScan |
|
||||||
public class MqttInboundConfig { |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private MqttPahoClientFactory mqttClientFactory; |
|
||||||
|
|
||||||
@Resource(name = ChannelName.INBOUND) |
|
||||||
private MessageChannel inboundChannel; |
|
||||||
|
|
||||||
private String clientId; |
|
||||||
|
|
||||||
/** |
|
||||||
* 入站适配器 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean(name = "adapter") |
|
||||||
public MessageProducerSupport mqttInbound() { |
|
||||||
MqttClientOptions options = MqttConfig.getBasicMqttClientOptions(); |
|
||||||
// 此处初始化的时候，默认订阅了配置文件中已经写好的topic
|
|
||||||
// 如果需要订阅多个，可以自己手动订阅，会写一个addTopic()进行添加订阅
|
|
||||||
clientId = options.getClientId() + "_consumer_" + System.currentTimeMillis(); |
|
||||||
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter( |
|
||||||
clientId, |
|
||||||
mqttClientFactory, |
|
||||||
options.getInboundTopic().split(",")); |
|
||||||
// System.out.println("每一次都会入站适配器吗？"+clientId);
|
|
||||||
DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(); |
|
||||||
// 统一是字节处理
|
|
||||||
converter.setPayloadAsBytes(true); |
|
||||||
// 设置消息转换器
|
|
||||||
adapter.setConverter(converter); |
|
||||||
// 设置qos（quality of service）
|
|
||||||
// 0：最多一次传输（消息会丢失），
|
|
||||||
// 1：至少一次传输（消息会重复），
|
|
||||||
// 2：只有当消息发送成功时才确认（消息不回丢，但延迟高）。
|
|
||||||
adapter.setQos(0); |
|
||||||
// 设置在接收已经订阅的主题信息后，发送给哪个通道，具体的发送方法需要翻上层的抽象类
|
|
||||||
adapter.setOutputChannel(inboundChannel); |
|
||||||
return adapter; |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 默认声明一个消息处理器，用于处理无效的消息 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean |
|
||||||
@ServiceActivator(inputChannel = ChannelName.DEFAULT_BOUND) |
|
||||||
public MessageHandler handler() { |
|
||||||
return message -> { |
|
||||||
log.info("The default channel does not handle messages." + |
|
||||||
"\nTopic: {}" + |
|
||||||
"\nPayload: {}", |
|
||||||
message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC), |
|
||||||
message.getPayload()); |
|
||||||
}; |
|
||||||
} |
|
||||||
|
|
||||||
public String getClientId() { |
|
||||||
return clientId; |
|
||||||
} |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,55 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.config; |
|
||||||
|
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import lombok.extern.slf4j.Slf4j; |
|
||||||
import org.springframework.context.annotation.Bean; |
|
||||||
import org.springframework.context.annotation.Configuration; |
|
||||||
import org.springframework.integration.channel.DirectChannel; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 声明所有通道的定义类 |
|
||||||
* @date 2024-10-29 16:23:32 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Configuration |
|
||||||
public class MqttMessageChannel { |
|
||||||
|
|
||||||
@Bean(name = ChannelName.OUTBOUND) |
|
||||||
public MessageChannel outboundChannel() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.INBOUND) |
|
||||||
public MessageChannel inboundChannel() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
/** |
|
||||||
* 事件主动上报通道 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean(name = ChannelName.EVENTS_UPLOAD_INBOUND) |
|
||||||
public MessageChannel eventsUploadInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.EVENTS_COLLECTION_INBOUND) |
|
||||||
public MessageChannel eventsCollectionInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.EVENTS_CONTROL_INBOUND) |
|
||||||
public MessageChannel eventsControlInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
@Bean(name = ChannelName.EVENTS_SEND_INBOUND) |
|
||||||
public MessageChannel eventsSendInbound() { |
|
||||||
return new DirectChannel(); |
|
||||||
} |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,51 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.config; |
|
||||||
|
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
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.IntegrationComponentScan; |
|
||||||
import org.springframework.integration.annotation.ServiceActivator; |
|
||||||
import org.springframework.integration.mqtt.core.MqttPahoClientFactory; |
|
||||||
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; |
|
||||||
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; |
|
||||||
import org.springframework.messaging.MessageHandler; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 入站配置 |
|
||||||
* @date 2024-10-29 16:22:10 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Configuration |
|
||||||
@IntegrationComponentScan |
|
||||||
public class MqttOutboundConfig { |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private MqttPahoClientFactory mqttClientFactory; |
|
||||||
|
|
||||||
/** |
|
||||||
* 默认声明一个出站处理器，用于处理无效的消息 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Bean |
|
||||||
@ServiceActivator(inputChannel = ChannelName.OUTBOUND) |
|
||||||
public MessageHandler mqttOutbound() { |
|
||||||
MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler( |
|
||||||
MqttConfig.getBasicMqttClientOptions().getClientId() + "_producer_" + System.currentTimeMillis(), |
|
||||||
mqttClientFactory); |
|
||||||
DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(); |
|
||||||
// use byte types uniformly
|
|
||||||
converter.setPayloadAsBytes(true); |
|
||||||
|
|
||||||
messageHandler.setAsync(true); |
|
||||||
messageHandler.setDefaultQos(0); |
|
||||||
messageHandler.setConverter(converter); |
|
||||||
return messageHandler; |
|
||||||
} |
|
||||||
|
|
||||||
|
|
||||||
} |
|
||||||
@ -1,62 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.handler; |
|
||||||
|
|
||||||
import com.mh.user.config.MHConfig; |
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import com.mh.user.constants.TopicEnum; |
|
||||||
import com.mh.user.utils.SpringContextUtils; |
|
||||||
import lombok.extern.slf4j.Slf4j; |
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||||
import org.springframework.integration.annotation.Router; |
|
||||||
import org.springframework.integration.mqtt.support.MqttHeaders; |
|
||||||
import org.springframework.integration.router.AbstractMessageRouter; |
|
||||||
import org.springframework.messaging.Message; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
import org.springframework.messaging.MessageHeaders; |
|
||||||
import org.springframework.stereotype.Component; |
|
||||||
|
|
||||||
import java.util.Collection; |
|
||||||
import java.util.Collections; |
|
||||||
import java.util.Objects; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 入站消息路由分发中心 |
|
||||||
* @date 2024-10-29 17:04:17 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Component |
|
||||||
public class InboundMessageRouter extends AbstractMessageRouter { |
|
||||||
|
|
||||||
/** 系统基础配置 */ |
|
||||||
@Autowired |
|
||||||
private MHConfig mHConfig; |
|
||||||
|
|
||||||
/** |
|
||||||
* 目前只需要这个方式，后期在拓展使用IntegrationFlow方式 |
|
||||||
* @param message |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
@Router(inputChannel = ChannelName.INBOUND) |
|
||||||
@Override |
|
||||||
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) { |
|
||||||
MessageHeaders headers = message.getHeaders(); |
|
||||||
String topic = Objects.requireNonNull(headers.get(MqttHeaders.RECEIVED_TOPIC)).toString(); |
|
||||||
// byte[] payload = (byte[]) message.getPayload();
|
|
||||||
// log.info("从当前主题 topic: {}, 接收到的消息：{}", topic, new String(payload));
|
|
||||||
// 判断当前主题是否是当前项目的,温湿度目前写死的
|
|
||||||
// if (!topic.startsWith(mHConfig.getName()) && !topic.contains("/temp")) {
|
|
||||||
// log.info("当前主题 topic: {} 不是当前项目的，直接丢弃", topic);
|
|
||||||
// return Collections.singleton((MessageChannel) SpringContextUtils.getBean(ChannelName.DEFAULT_BOUND));
|
|
||||||
// }
|
|
||||||
// 找到对应的主题消息通道
|
|
||||||
if (topic.contains("/temp")) { |
|
||||||
return Collections.singleton((MessageChannel) SpringContextUtils.getBean(ChannelName.EVENTS_UPLOAD_INBOUND)); |
|
||||||
} else { |
|
||||||
TopicEnum topicEnum = TopicEnum.find(mHConfig.getName() + "/", topic); |
|
||||||
MessageChannel bean = (MessageChannel) SpringContextUtils.getBean(topicEnum.getBeanName()); |
|
||||||
return Collections.singleton(bean); |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,49 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service; |
|
||||||
|
|
||||||
import org.springframework.messaging.MessageHeaders; |
|
||||||
|
|
||||||
import java.io.IOException; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 通道处理类 |
|
||||||
* @date 2024-11-05 11:30:26 |
|
||||||
*/ |
|
||||||
public interface IEventsService { |
|
||||||
|
|
||||||
/** |
|
||||||
* 处理主动上报 |
|
||||||
* @param receiver |
|
||||||
* @param headers |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
void handleInboundUpload(byte[] receiver, MessageHeaders headers) throws IOException; |
|
||||||
|
|
||||||
|
|
||||||
/** |
|
||||||
* 处理服务器主动采集上报 |
|
||||||
* @param receiver |
|
||||||
* @param headers |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
void handleInboundCollection(byte[] receiver, MessageHeaders headers) throws IOException; |
|
||||||
|
|
||||||
/** |
|
||||||
* 处理控制上报 |
|
||||||
* @param receiver |
|
||||||
* @param headers |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
void handleInboundControl(byte[] receiver, MessageHeaders headers) throws IOException; |
|
||||||
|
|
||||||
/** |
|
||||||
* 处理发送 |
|
||||||
* @param receiver |
|
||||||
* @param headers |
|
||||||
* @throws IOException |
|
||||||
*/ |
|
||||||
void handleInboundSend(byte[] receiver, MessageHeaders headers) throws IOException; |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,36 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt |
|
||||||
* @description 消息网关 |
|
||||||
* @date 2024-10-30 14:43:55 |
|
||||||
*/ |
|
||||||
//@Component
|
|
||||||
//@Configuration
|
|
||||||
//@MessagingGateway(defaultRequestChannel = ChannelName.OUTBOUND)
|
|
||||||
public interface IMqttGatewayService { |
|
||||||
|
|
||||||
/** |
|
||||||
* 发送消息 |
|
||||||
* @param topic |
|
||||||
* @param payload |
|
||||||
*/ |
|
||||||
void publish(String topic, String payload, int qos); |
|
||||||
|
|
||||||
/** |
|
||||||
* 发送消息 |
|
||||||
* @param topic |
|
||||||
* @param payload |
|
||||||
*/ |
|
||||||
void publish(String topic, byte[] payload); |
|
||||||
|
|
||||||
/** |
|
||||||
* 发送消息并带上qos |
|
||||||
* @param topic |
|
||||||
* @param payload |
|
||||||
* @param qos |
|
||||||
*/ |
|
||||||
void publish(String topic, byte[] payload, int qos); |
|
||||||
} |
|
||||||
@ -1,96 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service; |
|
||||||
|
|
||||||
import com.fasterxml.jackson.core.type.TypeReference; |
|
||||||
import com.mh.user.constants.CommonTopicResponse; |
|
||||||
import com.mh.user.constants.ServiceReply; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description Mqtt发送消息 |
|
||||||
* @date 2024-10-29 17:31:40 |
|
||||||
*/ |
|
||||||
public interface IMqttMsgSenderService { |
|
||||||
|
|
||||||
void publish(String topic, String pushMessage); |
|
||||||
|
|
||||||
/** |
|
||||||
* 发布消息 |
|
||||||
* @param topic target |
|
||||||
* @param response message |
|
||||||
*/ |
|
||||||
void publish(String topic, CommonTopicResponse response); |
|
||||||
|
|
||||||
/** |
|
||||||
* 使用qos发布消息 |
|
||||||
* |
|
||||||
* @param topic target |
|
||||||
* @param qos qos |
|
||||||
* @param response message |
|
||||||
*/ |
|
||||||
void publish(String topic, int qos, CommonTopicResponse response); |
|
||||||
|
|
||||||
/** |
|
||||||
* 发送消息并同时接收响应 |
|
||||||
* @param clazz |
|
||||||
* @param topic |
|
||||||
* @param response 通知启动是否成功 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
<T> T publishWithReply(Class<T> clazz, String topic, CommonTopicResponse response); |
|
||||||
|
|
||||||
/** |
|
||||||
* 发送消息并同时接收响应 |
|
||||||
* @param clazz |
|
||||||
* @param topic |
|
||||||
* @param response |
|
||||||
* @param retryTime |
|
||||||
* @param <T> |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
<T> T publishWithReply(Class<T> clazz, String topic, CommonTopicResponse response, int retryTime); |
|
||||||
|
|
||||||
/** |
|
||||||
* 专门用于发送服务消息 |
|
||||||
* @param clazz The generic class for ServiceReply. |
|
||||||
* @param sn |
|
||||||
* @param method |
|
||||||
* @param data |
|
||||||
* @param bid |
|
||||||
* @param <T> |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
<T> ServiceReply<T> publishServicesTopic(TypeReference<T> clazz, String sn, String method, Object data, String bid); |
|
||||||
|
|
||||||
/** |
|
||||||
* 仅用于为服务发送消息，不设置接收到的子类型 |
|
||||||
* @param sn |
|
||||||
* @param method |
|
||||||
* @param data |
|
||||||
* @param bid |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
ServiceReply publishServicesTopic(String sn, String method, Object data, String bid); |
|
||||||
|
|
||||||
/** |
|
||||||
* 专门用于发送服务消息 |
|
||||||
* @param clazz The generic class for ServiceReply. |
|
||||||
* @param sn |
|
||||||
* @param method |
|
||||||
* @param data |
|
||||||
* @param <T> |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
<T> ServiceReply<T> publishServicesTopic(TypeReference<T> clazz, String sn, String method, Object data); |
|
||||||
|
|
||||||
/** |
|
||||||
* 仅用于为服务发送消息，不设置接收到的子类型 |
|
||||||
* @param sn |
|
||||||
* @param method |
|
||||||
* @param data |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
ServiceReply publishServicesTopic(String sn, String method, Object data); |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,40 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service; |
|
||||||
|
|
||||||
import org.springframework.integration.mqtt.support.MqttHeaders; |
|
||||||
import org.springframework.messaging.handler.annotation.Header; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description mqtt订阅主题 |
|
||||||
* @date 2024-10-29 17:29:52 |
|
||||||
*/ |
|
||||||
public interface IMqttTopicService { |
|
||||||
|
|
||||||
/** |
|
||||||
* 订阅主题 |
|
||||||
* @param topic |
|
||||||
*/ |
|
||||||
void subscribe(@Header(MqttHeaders.TOPIC) String topic); |
|
||||||
|
|
||||||
/** |
|
||||||
* 订阅主题并设置qos |
|
||||||
* @param topic |
|
||||||
* @param qos |
|
||||||
*/ |
|
||||||
void subscribe(@Header(MqttHeaders.TOPIC) String topic, int qos); |
|
||||||
|
|
||||||
/** |
|
||||||
* 解绑主题 |
|
||||||
* @param topic |
|
||||||
*/ |
|
||||||
void unsubscribe(@Header(MqttHeaders.TOPIC) String topic); |
|
||||||
|
|
||||||
/** |
|
||||||
* 获取已订阅的主题 |
|
||||||
* @return |
|
||||||
*/ |
|
||||||
String[] getSubscribedTopics(); |
|
||||||
|
|
||||||
} |
|
||||||
@ -1,75 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service.impl; |
|
||||||
|
|
||||||
import com.alibaba.fastjson2.JSONObject; |
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper; |
|
||||||
import com.mh.user.constants.ChannelName; |
|
||||||
import com.mh.user.model.SanShiFengReceiver; |
|
||||||
import com.mh.user.service.mqtt.service.IEventsService; |
|
||||||
import io.netty.util.CharsetUtil; |
|
||||||
import lombok.extern.slf4j.Slf4j; |
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||||
import org.springframework.integration.annotation.ServiceActivator; |
|
||||||
import org.springframework.messaging.MessageHeaders; |
|
||||||
import org.springframework.stereotype.Service; |
|
||||||
|
|
||||||
import java.io.IOException; |
|
||||||
import java.util.Objects; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 通道消息处理 |
|
||||||
* @date 2024-11-05 11:42:30 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Service |
|
||||||
public class EventsServiceImpl implements IEventsService { |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private ObjectMapper mapper; |
|
||||||
|
|
||||||
@ServiceActivator(inputChannel = ChannelName.EVENTS_UPLOAD_INBOUND) |
|
||||||
@Override |
|
||||||
public void handleInboundUpload(byte[] receiver, MessageHeaders headers) { |
|
||||||
// 获取当前的主题
|
|
||||||
String topic = Objects.requireNonNull(headers.get("mqtt_receivedTopic")).toString(); |
|
||||||
handleInboundData(receiver, topic, "主动上报数据"); |
|
||||||
} |
|
||||||
|
|
||||||
@ServiceActivator(inputChannel = ChannelName.EVENTS_COLLECTION_INBOUND) |
|
||||||
@Override |
|
||||||
public void handleInboundCollection(byte[] receiver, MessageHeaders headers) { |
|
||||||
// 获取当前的主题
|
|
||||||
String topic = Objects.requireNonNull(headers.get("mqtt_receivedTopic")).toString(); |
|
||||||
handleInboundData(receiver, topic, "主动下发采集数据"); |
|
||||||
} |
|
||||||
|
|
||||||
@ServiceActivator(inputChannel = ChannelName.EVENTS_CONTROL_INBOUND) |
|
||||||
@Override |
|
||||||
public void handleInboundControl(byte[] receiver, MessageHeaders headers) { |
|
||||||
// 获取当前的主题
|
|
||||||
String topic = Objects.requireNonNull(headers.get("mqtt_receivedTopic")).toString(); |
|
||||||
handleInboundData(receiver, topic, "控制指令下发"); |
|
||||||
} |
|
||||||
|
|
||||||
@ServiceActivator(inputChannel = ChannelName.EVENTS_SEND_INBOUND) |
|
||||||
@Override |
|
||||||
public void handleInboundSend(byte[] receiver, MessageHeaders headers) throws IOException { |
|
||||||
String sendStr = new String(receiver, CharsetUtil.UTF_8); |
|
||||||
log.info("接收到控制指令下发=>{}", sendStr); |
|
||||||
} |
|
||||||
|
|
||||||
private void handleInboundData(byte[] receiver,String topic, String logMessage) { |
|
||||||
try { |
|
||||||
SanShiFengReceiver commonTopicReceiver = new SanShiFengReceiver(); |
|
||||||
commonTopicReceiver = mapper.readValue(receiver, SanShiFengReceiver.class); |
|
||||||
log.info("主题：{}，类型：{}: ，数据：{}", topic, logMessage, commonTopicReceiver.toString()); |
|
||||||
// 不使用消息队列，查询属于哪种设备类型，然后通过策略进行数据解析
|
|
||||||
} catch (IOException e) { |
|
||||||
log.error("处理数据时发生错误: ", e); |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
|
|
||||||
} |
|
||||||
@ -1,54 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service.impl; |
|
||||||
|
|
||||||
import com.mh.user.service.mqtt.service.IMqttGatewayService; |
|
||||||
import org.springframework.beans.factory.annotation.Qualifier; |
|
||||||
import org.springframework.integration.mqtt.support.MqttHeaders; |
|
||||||
import org.springframework.messaging.MessageChannel; |
|
||||||
import org.springframework.messaging.support.MessageBuilder; |
|
||||||
import org.springframework.stereotype.Service; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project EEMCS |
|
||||||
* @description 网关实现类 |
|
||||||
* @date 2025-02-07 08:44:55 |
|
||||||
*/ |
|
||||||
@Service |
|
||||||
public class MqttGatewayServiceImpl implements IMqttGatewayService { |
|
||||||
|
|
||||||
private final MessageChannel outboundChannel; |
|
||||||
|
|
||||||
public MqttGatewayServiceImpl(@Qualifier("outbound") MessageChannel outboundChannel) { |
|
||||||
this.outboundChannel = outboundChannel; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public synchronized void publish(String topic, String payload, int qos) { |
|
||||||
outboundChannel.send( |
|
||||||
MessageBuilder |
|
||||||
.withPayload(payload) |
|
||||||
.setHeader(MqttHeaders.TOPIC, topic) |
|
||||||
.setHeader(MqttHeaders.QOS, qos) |
|
||||||
.build()); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void publish(String topic, byte[] payload) { |
|
||||||
outboundChannel.send( |
|
||||||
MessageBuilder |
|
||||||
.withPayload(payload) |
|
||||||
.setHeader(MqttHeaders.TOPIC, topic) |
|
||||||
.build()); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void publish(String topic, byte[] payload, int qos) { |
|
||||||
outboundChannel.send( |
|
||||||
MessageBuilder |
|
||||||
.withPayload(payload) |
|
||||||
.setHeader(MqttHeaders.TOPIC, topic) |
|
||||||
.setHeader(MqttHeaders.QOS, qos) |
|
||||||
.build()); |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,142 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service.impl; |
|
||||||
|
|
||||||
import com.fasterxml.jackson.core.JsonProcessingException; |
|
||||||
import com.fasterxml.jackson.core.type.TypeReference; |
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper; |
|
||||||
import com.mh.user.constants.CommonTopicResponse; |
|
||||||
import com.mh.user.constants.ServiceReply; |
|
||||||
import com.mh.user.constants.TopicConst; |
|
||||||
import com.mh.user.service.mqtt.service.IMqttGatewayService; |
|
||||||
import com.mh.user.service.mqtt.service.IMqttMsgSenderService; |
|
||||||
import lombok.extern.slf4j.Slf4j; |
|
||||||
import org.springframework.beans.factory.annotation.Autowired; |
|
||||||
import org.springframework.stereotype.Service; |
|
||||||
import org.springframework.util.StringUtils; |
|
||||||
|
|
||||||
import java.util.Objects; |
|
||||||
import java.util.UUID; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description mqtt不同类型发送消息 |
|
||||||
* @date 2024-10-30 14:30:03 |
|
||||||
*/ |
|
||||||
@Slf4j |
|
||||||
@Service |
|
||||||
public class MqttMsgSenderServiceImpl implements IMqttMsgSenderService { |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private IMqttGatewayService mqttGatewayService; |
|
||||||
|
|
||||||
@Autowired |
|
||||||
private ObjectMapper mapper; |
|
||||||
|
|
||||||
|
|
||||||
/** |
|
||||||
* 发布，默认qos为0，非持久化 |
|
||||||
* |
|
||||||
* @param pushMessage |
|
||||||
* @param topic |
|
||||||
*/ |
|
||||||
@Override |
|
||||||
public void publish(String topic, String pushMessage) { |
|
||||||
synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
|
|
||||||
try { |
|
||||||
mqttGatewayService.publish(topic, pushMessage, 0); |
|
||||||
log.info("发送主题：{}，消息：{}", topic, pushMessage); |
|
||||||
} catch (Exception e) { |
|
||||||
log.error("发送主题异常：{}，消息：{}", topic, pushMessage, e); |
|
||||||
throw new RuntimeException(e); |
|
||||||
} |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void publish(String topic, CommonTopicResponse response) { |
|
||||||
this.publish(topic, 1, response); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void publish(String topic, int qos, CommonTopicResponse response) { |
|
||||||
try { |
|
||||||
log.info("发送主题：{}，消息：{}", topic, response.toString()); |
|
||||||
mqttGatewayService.publish(topic, mapper.writeValueAsBytes(response), qos); |
|
||||||
} catch (JsonProcessingException e) { |
|
||||||
log.error("发送主题：{}，消息：{}", topic, response.toString(), e); |
|
||||||
throw new RuntimeException(e); |
|
||||||
} |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public <T> T publishWithReply(Class<T> clazz, String topic, CommonTopicResponse response) { |
|
||||||
return this.publishWithReply(clazz, topic, response, 2); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public <T> T publishWithReply(Class<T> clazz, String topic, CommonTopicResponse response, int retryTime) { |
|
||||||
// AtomicInteger time = new AtomicInteger(0);
|
|
||||||
// // Retry three times
|
|
||||||
// while (time.getAndIncrement() <= retryTime) {
|
|
||||||
// this.publish(topic, response);
|
|
||||||
//
|
|
||||||
// Chan<CommonTopicReceiver<T>> chan = Chan.getInstance();
|
|
||||||
// // If the message is not received in 0.5 seconds then resend it again.
|
|
||||||
// CommonTopicReceiver<T> receiver = chan.get(response.getTid());
|
|
||||||
//
|
|
||||||
// // Need to match tid and bid.
|
|
||||||
// if (Objects.nonNull(receiver) && receiver.getTid().equals(response.getTid()) &&
|
|
||||||
// receiver.getBid().equals(response.getBid())) {
|
|
||||||
// if (clazz.isAssignableFrom(receiver.getData().getClass())) {
|
|
||||||
// return receiver.getData();
|
|
||||||
// }
|
|
||||||
// throw new TypeMismatchException(receiver.getData(), clazz);
|
|
||||||
// }
|
|
||||||
// // It must be guaranteed that the tid and bid of each message are different.
|
|
||||||
// response.setBid(UUID.randomUUID().toString());
|
|
||||||
// response.setTid(UUID.randomUUID().toString());
|
|
||||||
// }
|
|
||||||
// throw new RuntimeException("没有消息接收到");
|
|
||||||
return null; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public <T> ServiceReply<T> publishServicesTopic(TypeReference<T> clazz, String sn, String method, Object data, String bid) { |
|
||||||
String topic = TopicConst.THING_MODEL_PRE + TopicConst.PRODUCT + sn + TopicConst.SERVICES_SUF; |
|
||||||
ServiceReply reply = this.publishWithReply(ServiceReply.class, topic, |
|
||||||
CommonTopicResponse.builder() |
|
||||||
.tid(UUID.randomUUID().toString()) |
|
||||||
.bid(StringUtils.hasText(bid) ? bid : UUID.randomUUID().toString()) |
|
||||||
.timestamp(System.currentTimeMillis()) |
|
||||||
.method(method) |
|
||||||
.data(Objects.requireNonNull(data, "")) |
|
||||||
.build()); |
|
||||||
if (Objects.isNull(clazz)) { |
|
||||||
return reply; |
|
||||||
} |
|
||||||
// put together in "output"
|
|
||||||
if (Objects.nonNull(reply.getInfo())) { |
|
||||||
reply.setOutput(mapper.convertValue(reply.getInfo(), clazz)); |
|
||||||
} |
|
||||||
if (Objects.nonNull(reply.getOutput())) { |
|
||||||
reply.setOutput(mapper.convertValue(reply.getOutput(), clazz)); |
|
||||||
} |
|
||||||
return reply; |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public ServiceReply publishServicesTopic(String sn, String method, Object data, String bid) { |
|
||||||
return this.publishServicesTopic(null, sn, method, data, bid); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public <T> ServiceReply<T> publishServicesTopic(TypeReference<T> clazz, String sn, String method, Object data) { |
|
||||||
return this.publishServicesTopic(clazz, sn, method, data, null); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public ServiceReply publishServicesTopic(String sn, String method, Object data) { |
|
||||||
return this.publishServicesTopic(null, sn, method, data, null); |
|
||||||
} |
|
||||||
} |
|
||||||
@ -1,40 +0,0 @@ |
|||||||
package com.mh.user.service.mqtt.service.impl; |
|
||||||
|
|
||||||
import com.mh.framework.mqtt.service.IMqttTopicService; |
|
||||||
import jakarta.annotation.Resource; |
|
||||||
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; |
|
||||||
import org.springframework.stereotype.Service; |
|
||||||
|
|
||||||
/** |
|
||||||
* @author LJF |
|
||||||
* @version 1.0 |
|
||||||
* @project springboot-mqtt-demo |
|
||||||
* @description 订阅主题实现类 |
|
||||||
* @date 2024-10-29 17:36:31 |
|
||||||
*/ |
|
||||||
@Service |
|
||||||
public class MqttTopicServiceImpl implements IMqttTopicService { |
|
||||||
|
|
||||||
@Resource |
|
||||||
private MqttPahoMessageDrivenChannelAdapter adapter; |
|
||||||
|
|
||||||
@Override |
|
||||||
public void subscribe(String topic) { |
|
||||||
adapter.addTopic(topic); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void subscribe(String topic, int qos) { |
|
||||||
adapter.addTopic(topic, qos); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public void unsubscribe(String topic) { |
|
||||||
adapter.removeTopic(topic); |
|
||||||
} |
|
||||||
|
|
||||||
@Override |
|
||||||
public String[] getSubscribedTopics() { |
|
||||||
return adapter.getTopic(); |
|
||||||
} |
|
||||||
} |
|
||||||
Loading…
Reference in new issue