47 changed files with 1931 additions and 30 deletions
@ -0,0 +1,122 @@ |
|||||||
|
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"; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,59 @@ |
|||||||
|
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"; |
||||||
|
} |
||||||
@ -0,0 +1,36 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
@ -0,0 +1,108 @@ |
|||||||
|
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 + '\'' + |
||||||
|
'}'; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,33 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,16 @@ |
|||||||
|
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 |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,20 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
@ -0,0 +1,31 @@ |
|||||||
|
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"; |
||||||
|
} |
||||||
@ -0,0 +1,60 @@ |
|||||||
|
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); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,48 @@ |
|||||||
|
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; |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,32 @@ |
|||||||
|
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); |
||||||
|
} |
||||||
@ -0,0 +1,30 @@ |
|||||||
|
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; |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,29 @@ |
|||||||
|
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; |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,26 @@ |
|||||||
|
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); |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,61 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,99 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,91 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,55 @@ |
|||||||
|
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(); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,51 @@ |
|||||||
|
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; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,62 @@ |
|||||||
|
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); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,49 @@ |
|||||||
|
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; |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,36 @@ |
|||||||
|
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); |
||||||
|
} |
||||||
@ -0,0 +1,96 @@ |
|||||||
|
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); |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,40 @@ |
|||||||
|
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(); |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,75 @@ |
|||||||
|
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); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,54 @@ |
|||||||
|
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()); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,142 @@ |
|||||||
|
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); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,40 @@ |
|||||||
|
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