1、在RuoYi-Vue项目的superVisualizationSys-common模块下的pom.xml加入jar包依赖
org.springframework.boot spring-boot-starter-integration org.springframework.integration spring-integration-stream org.springframework.integration spring-integration-mqtt 2、在application.yml添加相关配置，加在Spring配置下
# Spring配置spring: # mqtt mqtt: username: admin # 用户名 password: public # 密码 hostUrl: tcp:///broker.emqx.io:1883 # tcp://ip:端口 clientId: clientId # 客户端id defaultTopic: topic,topic1,test # 订阅主题 timeout: 100 # 超时时间 （单位：秒） keepalive: 60 # 心跳 （单位：秒） enabled: true # 是否使用mqtt功能
3、在superVisualizationSys-common\src\main\java\com\superVisualization\common\utils目录下新建mqtt文件夹，添加以下三个文件
MqttConfig.java
package com.superVisualization.common.utils.mqtt;import com.superVisualization.common.utils.StringUtils;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.context.properties.ConfigurationProperties;import org.springframework.context.annotation.Bean;import org.springframework.stereotype.Component;@Component@ConfigurationProperties("spring.mqtt")public class MqttConfig { @Autowired private MqttPushClient mqttPushClient; private String username; private String password; private String hostUrl; private String clientId; private String defaultTopic; private int timeout; private int keepalive; private boolean enabled; 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 getHostUrl() { return hostUrl; } public void setHostUrl(String hostUrl) { this.hostUrl = hostUrl; } public String getClientId() { return clientId; } public void setClientId(String clientId) { this.clientId = clientId; } public String getDefaultTopic() { return defaultTopic; } public void setDefaultTopic(String defaultTopic) { this.defaultTopic = defaultTopic; } public int getTimeout() { return timeout; } public void setTimeout(int timeout) { this.timeout = timeout; } public int getKeepalive() { return keepalive; } public void setKeepalive(int keepalive) { this.keepalive = keepalive; } public boolean isEnabled() { return enabled; } public void setEnabled(boolean enabled) { this.enabled = enabled; } @Bean public MqttPushClient getMqttPushClient() { if(enabled == true){ String mqtt_topic[] = StringUtils.split(defaultTopic, ","); mqttPushClient.connect(hostUrl, clientId, username, password, timeout, keepalive);//连接 for(int i=0; iMqttPushClient.java
package com.superVisualization.common.utils.mqtt;import com.superVisualization.common.core.domain.AjaxResult;import org.eclipse.paho.client.mqttv3.*;import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import static com.superVisualization.common.core.domain.AjaxResult.error;import static com.superVisualization.common.core.domain.AjaxResult.success;@Componentpublic class MqttPushClient { private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class); @Autowired private PushCallback pushCallback; private static MqttClient client; private static MqttClient getClient() { return client; } private static void setClient(MqttClient client) { MqttPushClient.client = client; } public void connect(String host, String clientID, String username, String password, int timeout, int keepalive) { MqttClient client; try { client = new MqttClient(host, clientID, new MemoryPersistence()); MqttConnectOptions options = new MqttConnectOptions(); options.setCleanSession(true); options.setUserName(username); options.setPassword(password.toCharArray()); options.setConnectionTimeout(timeout); options.setKeepAliveInterval(keepalive); MqttPushClient.setClient(client); try { client.setCallback(pushCallback); client.connect(options); } catch (Exception e) { e.printStackTrace(); } } catch (Exception e) { e.printStackTrace(); } } public AjaxResult publish(int qos, boolean retained, String topic, String pushMessage) { MqttMessage message = new MqttMessage(); message.setQos(qos); message.setRetained(retained); message.setPayload(pushMessage.getBytes()); MqttTopic mTopic = MqttPushClient.getClient().getTopic(topic); if (null == mTopic) { logger.error("topic not exist"); } MqttDeliveryToken token; try { token = mTopic.publish(message); token.waitForCompletion(); return success(); } catch (MqttPersistenceException e) { e.printStackTrace(); return error(); } catch (MqttException e) { e.printStackTrace(); return error(); } } public void subscribe(String topic, int qos) { logger.info("开始订阅主题" + topic); try { MqttPushClient.getClient().subscribe(topic, qos); } catch (MqttException e) { e.printStackTrace(); } }}PushCallback.java
package com.superVisualization.common.utils.mqtt;import com.alibaba.fastjson2.JSONObject;import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;import org.eclipse.paho.client.mqttv3.MqttCallback;import org.eclipse.paho.client.mqttv3.MqttClient;import org.eclipse.paho.client.mqttv3.MqttMessage;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class PushCallback implements MqttCallback { private static final Logger logger = LoggerFactory.getLogger(MqttPushClient.class); @Autowired private MqttConfig mqttConfig; private static MqttClient client; private static String _topic; private static String _qos; private static String _msg; @Override public void connectionLost(Throwable throwable) { // 连接丢失后，一般在这里面进行重连 logger.info("连接断开，可以做重连"); if (client == null || !client.isConnected()) { mqttConfig.getMqttPushClient(); } } @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { // subscribe后得到的消息会执行到这里面 logger.info("接收消息主题 : " + topic); logger.info("接收消息Qos : " + mqttMessage.getQos()); logger.info("接收消息内容 : " + new String(mqttMessage.getPayload())); _topic = topic; _qos = mqttMessage.getQos()+""; _msg = new String(mqttMessage.getPayload()); } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { logger.info("deliveryComplete---------" + iMqttDeliveryToken.isComplete()); } //别的Controller层会调用这个方法来 获取 接收到的硬件数据 public String receive() { JSONObject jsonObject = new JSONObject(); jsonObject.put("topic", _topic); jsonObject.put("qos", _qos); jsonObject.put("msg", _msg); return jsonObject.toString(); }}注：项目报红的地方可以按Alt+Enter键导包
二、下载安装emqx服务
1. 在windows上安装EMQX环境，操作系统是WINDOWS10系统，内存需2G以上,在官网上下载EMQX服务 (https://www.emqx.io)
2. 把下载的文件拷贝到工作目录下，可以新建一个文件夹，后续启动EMQX服务就在这个文件启动, 解压zip文件。
3. 先在etc目录下修改白名单，打开etc目录，找到acl.conf文件。
4. 修改后的文件如下，修改后文件记得保存
5. 在回到上一级目录，进入bin目录，电脑在左下角输入框内输入“cmd“，最好直接选择以管理员权限运行
6. 通过cmd指令进入刚才解压的emqx文件的bin目录下，复制bin文件的目录，在cmd命令行输入pushd + 复制的目录
例如：
pushd D:\Tools\emqx-4.4.16-otp24.3.4.6-windows-amd64\emqx\bin 敲击回车。
7. 执行 emqx.cmd start 回车，windows上start没返回，打开浏览器输入： localhost:18083
8. 出现如下界面便是EMQX服务已经启动，EMQX初始的用户名：admin 密码：public 。
9. 在实际使用中，本地IP最好设置成固定IP，可通过电脑的IP访问EMQX服务器,也可以使用默认IP 127.0.0.1
10. 在侧边栏的Clients窗口，在这里界面可以看到客户端设备连接的信息。
11. 以上EMQX服务搭建完成，目前针对订阅未进行白名单设置，即安装了EMQX无需改动参数，把MQTT服务器指向本地的地址就可以正常订阅发布消息。
备注：
其他参数如果有兴趣可以自行研究下。
配置emqx的参数在emqx--etc目录下的emqx.conf。
配置白名单或者订阅主题的在acl.conf。
三、下载MQTT X测试工具, 进行安装
1. 打开MQTT X，新建连接，输入名称，点击右上角(connect)连接
2. 模拟客户端订阅服务器主题发送信息
主题(topic)
3. 模拟客户端订阅服务器主题发送信息
主题(topic1)
4. 后续需要添加订阅主题,依次在(defaultTopic)后添加即可
-
EMQX启动命令说明
备注：
emqx的命令 ：
启动 ：emqx.cmd start
重启 ：emqx.cmd restart
停止 ：emqx.cmd stop
有疑问的同学可以留言评论。
后续会发布Linux版本实现EMQX服务器,实现消息的订阅发布 和 数据存储MySQL等.
觉得博主可以的,可以点赞加关注订阅 再走噢
点赞关注不迷路,博主带你学技术
来源地址：https://blog.csdn.net/xu275321143/article/details/129982081
