spring boot 添加 mqtt 收发消息保存消息等
<dependency>
<groupid>org.eclipse.paho</groupid>
<artifactid>org.eclipse.paho.client.mqttv3</artifactid>
<version>1.2.3</version>
</dependency>
mqtt:
urls: tcp://127.0.0.1:1883
clientId: WebKn9wu
username: king
password: 12341234
completionTimeout: 30000
/**
* Mqtt 属性配置类
*/
@Component
@Configuration
@Setter
@Getter
public class MqttConfig {
@Value("${spring.mqtt.username}")
private String username;
@Value("${spring.mqtt.password}")
private String password;
@Value("${spring.mqtt.urls}")
private String urls;
@Value("${spring.mqtt.clientId}")
private String clientId;
/** automaticReconnection 是否自动重连 默认false */
private boolean automaticReconnection = true;
/**
* 连接超时
*/
@Value("${spring.mqtt.completionTimeout}")
private int completionTimeout;
}
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 javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
/**
* Mqtt 创建连接 消息发布 订阅
*/
@Component
public class MyMqttClient {
private static final Logger logger = LoggerFactory.getLogger(MyMqttClient.class);
@Autowired
private MqttConfig config;
@Autowired
private PushCallback pushCallback;
@Autowired
private MqttLogService mqttLogService;
private org.eclipse.paho.client.mqttv3.MqttClient client;
private static MemoryPersistence memoryPersistence = null;
/** options MQTT Cleint连接配置 */
private MqttConnectOptions options;
public org.eclipse.paho.client.mqttv3.MqttClient getClient() {
return client;
}
public void setClient(org.eclipse.paho.client.mqttv3.MqttClient client) {
this.client = client;
}
/**
* connect : 客户端连接
*/
@PostConstruct
public void connect() {
try {
// 设置持久化方式
memoryPersistence = new MemoryPersistence();
this.client = new org.eclipse.paho.client.mqttv3.MqttClient(config.getUrls(), config.getClientId(), memoryPersistence);
options = new MqttConnectOptions();
// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
// 重连清空session
options.setCleanSession(true);
// 设置连接的用户名
options.setUserName(config.getUsername());
// 设置连接的密码
options.setPassword(config.getPassword().toCharArray());
// 设置超时时间 单位为秒
options.setConnectionTimeout(config.getCompletionTimeout());
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
options.setKeepAliveInterval(30);
// 是否重连
options.setAutomaticReconnect(true);
// options.setMaxInflight(config.getMaxInflight());
this.setClient(client);
client.setCallback(pushCallback);
// 如果是new的回调 可以把 业务接口传过去
// client.setCallback(new DefaultCallback(this, mqttLogService));
if (null !=client){
client.connect(options);
}
} catch (MqttException e) {
logger.error(e.getCause().getLocalizedMessage());
e.printStackTrace();
}
}
/**
* publish : 发布
*
* @param qos 连接方式
* @param retained 是否保留
* @param topic 主题
* @param pushMessage 消息体
*/
public void 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 = this.getClient().getTopic(topic);
if (null == mTopic) {
logger.info("topic not exist");
}
MqttDeliveryToken token;
try {
token = mTopic.publish(message);
token.waitForCompletion();
} catch (MqttPersistenceException e) {
e.printStackTrace();
} catch (MqttException e) {
e.printStackTrace();
}
}
/**
* 订阅某个主题
* @param topic 主题
* @param qos 消息质量 0 1 2
*/
public void subscribe(String topic, int qos) {
if (null != client && client.isConnected()) {
logger.info("开始订阅主题" + topic);
try {
this.getClient().subscribe(topic, qos);
} catch (MqttException e) {
e.printStackTrace();
}
}else {
System.out.println("mqttClient is error");
}
}
/**
* 取消订阅
* @param topic 要取消的主题
*/
public void cleanTopic(String topic) {
if (null != client && !client.isConnected()) {
try {
client.unsubscribe(topic);
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} else {
System.out.println("mqttClient is error");
}
}
/**关闭连接*/
@PreDestroy
public void closeConnect() {
//关闭存储方式
if (null != memoryPersistence) {
try {
memoryPersistence.close();
} catch (MqttPersistenceException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
} else {
logger.error("memoryPersistence is null");
}
// 关闭连接
if (null != client) {
if (client.isConnected()) {
try {
client.disconnect();
client.close();
} catch (MqttException e) {
// TODO Auto-generated catch block
e.printStackTrace();
logger.error("关闭连接异常={}",e.getMessage());
}
} else {
logger.error("关闭连接异常={}","mqttClient is not connect");
}
} else {
logger.error("关闭连接异常={}","mqttClient is null");
}
}
}
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/***
* Mqtt 消息返回指定接口
*/
@Slf4j
@Component
public class PushCallback implements MqttCallbackExtended {
private static final Logger logger = LoggerFactory.getLogger(PushCallback.class);
@Autowired
private MyMqttClient client;
@Autowired
private MqttConfig mqttConfig;
// 这是保存数据的接口
@Autowired
private MqttLogService mqttLogService;
/**
* 丢失了对服务端的连接后触发的回调
* @param throwable
*/
@Override
public void connectionLost(Throwable throwable) {
new Thread(() -> {
logger.info("[MQTT] 连接断开，1s之后尝试重连...", throwable);
boolean reconnecting = false;
for (int i = 1; i < 5; i++) {
try {
if (client.getClient().isConnected()) {
break;
}
Thread.sleep(1000);
boolean needReconnect = !mqttConfig.isAutomaticReconnection() && !reconnecting && !client.getClient().isConnected();
if (needReconnect) {
logger.info("开始重连...");
client.getClient().reconnect();
reconnecting = true;
}
} catch (Exception e) {
logger.info("mqtt重连失败,继续重连,reason:" + e.getMessage(), e);
continue;
}
}
}).start();
}
/**
* 应用收到消息后触发的回调
* @param topic
* @param mqttMessage
* @throws Exception
*/
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
log.info("=====接收消息主题 : " + topic);
String msg = new String(mqttMessage.getPayload());
log.info("=====接收消息内容 : " + msg);
String[] tsplit = topic.split("/");
// 调用 MqttLogService 内的存储方法
}
/**
* 消息发布者消息发布完成产生的回调
* @param iMqttDeliveryToken
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("=====deliveryComplete---------" + iMqttDeliveryToken.isComplete());
}
/**
* 连接完成 触发回调
* @param b
* @param s
*/
@Override
public void connectComplete(boolean b, String s) {
// 订阅间隔自动上报数据
client.subscribe("$king/devices/xxx/#", 0);
// 订阅设备信息
client.subscribe("$king/devices/xxx/#", 0);
log.info("mqtt连接成功，客户端ID：" + mqttConfig.getUrls());
// 对连接成功 保存日志
}
}
第二种 Callback
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
/***
* Mqtt 消息返回指定接口
*/
@Slf4j
@Component
public class DefaultCallback implements MqttCallbackExtended {
private static final Logger logger = LoggerFactory.getLogger(DefaultCallback.class);
private MqttLogService mqttLogService;
private MyMqttClient client;
public DefaultCallback(){
}
public DefaultCallback(MyMqttClient pushClient, MqttLogService service) {
super();
this.client = pushClient;
this.mqttLogService = service;
}
/**
* 丢失了对服务端的连接后触发的回调
* @param throwable
*/
@Override
public void connectionLost(Throwable throwable) {
new Thread(() -> {
logger.info("[MQTT] 连接断开，1s之后尝试重连...", throwable);
MqttClient mqttClient = client.getClient();
boolean reconnecting = false;
for (int i = 1; i < 5; i++) {
try {
if (mqttClient.isConnected()) {
break;
}
Thread.sleep(1000);
boolean needReconnect = !client.config.isAutomaticReconnection() && !reconnecting && !mqttClient.isConnected();
if (needReconnect) {
logger.info("开始重连...");
mqttClient.reconnect();
reconnecting = true;
}
} catch (Exception e) {
logger.info("mqtt重连失败,继续重连,reason:" + e.getMessage(), e);
continue;
}
}
}).start();
}
/**
* 应用收到消息后触发的回调
* @param topic
* @param mqttMessage
* @throws Exception
*/
@Override
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
log.info("=====接收消息主题 : " + topic);
String msg = new String(mqttMessage.getPayload());
log.info("=====接收消息内容 : " + msg);
String[] tsplit = topic.split("/");
// mqttLogService.saveMqtt(mqttLog);
log.info("-----对数据进行持久化");
}
/**
* 消息发布者消息发布完成产生的回调
* @param iMqttDeliveryToken
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
log.info("=====deliveryComplete---------" + iMqttDeliveryToken.isComplete());
}
/**
* 连接完成 触发回调
* @param b
* @param s
*/
@Override
public void connectComplete(boolean b, String s) {
// 订阅间隔自动上报数据
client.subscribe("$king/devices/xxx/#", 0);
// 订阅设备信息
client.subscribe("$king/devices/xxx/#", 0);
log.info("mqtt连接成功，客户端ID：" + client.config.getUrls());
// 对连接成功 保存日志
}
}
Original: https://www.cnblogs.com/KingPingyue/p/15954751.html
Author: 汤姆·希德勒斯顿
Title: 记录一下SpringBoot+MQTT的使用
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/592479/
转载文章受原作者版权保护。转载请注明原作者出处！