From 71e051cddfba05a4e7876a92650cc8d7b5fabf28 Mon Sep 17 00:00:00 2001
From: huliguo <2023611923@qq.com>
Date: 星期五, 23 五月 2025 20:59:58 +0800
Subject: [PATCH] Merge branch 'dev' of http://120.76.84.145:10101/gitblit/r/java/mx_charging_pile into dev
---
ruoyi-service/ruoyi-jianguan/src/main/java/com/ruoyi/jianguan/mqtt/util/MqttPushUtil.java | 169 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 169 insertions(+), 0 deletions(-)
diff --git a/ruoyi-service/ruoyi-jianguan/src/main/java/com/ruoyi/jianguan/mqtt/util/MqttPushUtil.java b/ruoyi-service/ruoyi-jianguan/src/main/java/com/ruoyi/jianguan/mqtt/util/MqttPushUtil.java
new file mode 100644
index 0000000..92f360b
--- /dev/null
+++ b/ruoyi-service/ruoyi-jianguan/src/main/java/com/ruoyi/jianguan/mqtt/util/MqttPushUtil.java
@@ -0,0 +1,169 @@
+package com.ruoyi.jianguan.mqtt.util;
+
+
+import com.ruoyi.common.core.domain.R;
+import com.ruoyi.jianguan.mqtt.callback.PushCallback;
+import com.ruoyi.jianguan.mqtt.client.ClientMQTT;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class MqttPushUtil {
+
+// private final static String HOST = "tcp://127.0.0.1:9882";
+// private final static String CLIENT_ID = "mqttx_3267186711111fdsasdsa";
+// private final static String USER_NAME = "admin";
+// private final static String PASS_WORD = "mingxingdianli123";
+ private final static String HOST = "tcp://183.220.129.95:1883";
+ private final static String CLIENT_ID = "mqttx_3267186711111mxcdszyun";
+ private final static String USER_NAME = "mxEmqx";
+ private final static String PASS_WORD = "admin";
+ private final static Integer TIME_OUT = 30000;
+ private final static Integer KEEP_ALIVE = 60000;
+ private MqttClient mqttClient;
+
+ /**
+ * 推送充电桩数据
+ * @param topic
+ * @param message
+ * @return
+ */
+ public R<String> pushChargePileData(String topic, String message){
+ try {
+ //发布消息
+ this.publish(topic,message);
+ return R.ok("["+topic+"]-推送成功;内容为:["+message+"]");
+ } catch (MqttException e) {
+ return R.fail("推送失败:["+e.getMessage()+"]");
+ }
+ }
+
+ /**
+ * 测试订阅消息
+ */
+// public static void main(String[] args) throws Exception {
+// MqttPushUtil mqttConnect = new MqttPushUtil();
+// mqttConnect.start();
+// //订阅消息
+// mqttConnect.publish("/cmt/IoT/pub/8/6700/status/silent","message11消息");
+// }
+
+ @Bean
+ public MqttPushUtil getMqttPushClient() {
+ //连接至mqtt服务器，获取mqtt连接
+ this.start();
+ return this;
+ }
+ /**
+ * 初始化mqtt链接
+ * @throws MqttException
+ */
+ public void start(){
+ try {
+ // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
+ mqttClient = new MqttClient(HOST, CLIENT_ID + System.currentTimeMillis(), new MemoryPersistence());
+ // MQTT的连接设置
+ MqttConnectOptions options = new MqttConnectOptions();
+ options.setUserName(USER_NAME);
+ options.setPassword(PASS_WORD.toCharArray());
+ // 设置超时时间 单位为秒
+ options.setConnectionTimeout(TIME_OUT);///默认：30
+ // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
+ options.setCleanSession(false);//默认：true
+ // 设置断开后重新连接(设置为true时将启用自动重新连接)
+ options.setAutomaticReconnect(true);//默认：false
+ // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
+ options.setKeepAliveInterval(KEEP_ALIVE);//默认：60
+ // 设置回调
+ mqttClient.setCallback(new PushCallback());
+ mqttClient.connect(options);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * 自定义mqtt连接
+ * @param host
+ * @param clientId
+ * @param userName
+ * @param passWord
+ * @param connectionTimeout
+ * @param cleanSession
+ * @param automaticReconnect
+ * @param keepAliveInterval
+ * @param mqttCallback
+ * @throws MqttException
+ */
+ public void start(String host,String clientId, String userName, String passWord,
+ int connectionTimeout, boolean cleanSession,boolean automaticReconnect,
+ int keepAliveInterval,MqttCallback mqttCallback) throws MqttException {
+ // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
+ mqttClient = new MqttClient(host, clientId + System.currentTimeMillis(), new MemoryPersistence());
+ // MQTT的连接设置
+ MqttConnectOptions options = new MqttConnectOptions();
+ options.setUserName(userName);
+ options.setPassword(passWord.toCharArray());
+ // 设置超时时间 单位为秒
+ options.setConnectionTimeout(connectionTimeout);///默认：30
+ // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
+ options.setCleanSession(cleanSession);//默认：true
+ // 设置断开后重新连接(设置为true时将启用自动重新连接)
+ options.setAutomaticReconnect(automaticReconnect);//默认：false
+ // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
+ options.setKeepAliveInterval(keepAliveInterval);//默认：60
+ // 设置回调
+ mqttClient.setCallback(mqttCallback);
+ mqttClient.connect(options);
+ }
+
+ /**
+ * 发布，默认qos为0，非持久化
+ */
+ public void publish(String topic, String message1) throws MqttException {
+ publish(0, false, topic, message1);
+ }
+
+ /**
+ * 发布
+ */
+ public void publish(int qos, boolean retained, String topic, String message1) throws MqttException {
+ MqttMessage message = new MqttMessage();
+ message.setQos(qos);
+ message.setRetained(retained);
+ message.setPayload(message1.getBytes());
+ MqttTopic mTopic = mqttClient.getTopic(topic);
+ if (null == mTopic) {
+ System.err.println("topic not exist");
+ log.error("topic not exist");
+ }
+ MqttDeliveryToken token;
+ token = mTopic.publish(message);
+ token.waitForCompletion();
+ }
+
+ /**
+ * 订阅某个主题，qos默认为0
+ */
+ public void subscribe(String topic) {
+ subscribe(topic, 0);
+ }
+
+ /**
+ * 订阅某个主题
+ */
+ public void subscribe(String topic, int qos) {
+ try {
+ mqttClient.subscribe(topic, qos);
+ } catch (MqttException e) {
+ e.printStackTrace();
+ }
+ }
+
+
+}
+
--
Gitblit v1.7.1