SpringBoot 结合官网对MQTT消息队列整合
首先是 maven Pom 的引入 Mqtt Client
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.2</version>
</dependency>
其次编写一个Mqtt回调方法类 去实现 MqttCallback 接口 用于消息的接收
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import java.util.Base64;
/**
* @author caicai
*
*
* 消息回调显示
*/
@Slf4j
public class PushCallback implements MqttCallback {
/**
* 连接丢失
* @param throwable
*/
@Override
public void connectionLost(Throwable throwable) {
}
/**
* 消息到达
* @param s
* @param mqttMessage
* @throws Exception
*/
@Override
public void messageArrived(java.lang.String s, MqttMessage mqttMessage) throws Exception {
log.info("com.kxcd.testkfka.mqtt.tools.PushCallback.messageArrived===>接收到topic {} ",s);
byte[] payload = mqttMessage.getPayload();
String s1 = Base64.getEncoder().encodeToString(payload);
log.info("PushCallback.messageArrived===>接收到消息内容 {} ",mqttMessage);
}
/**
* 消息发送完成
* @param iMqttDeliveryToken
*/
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
}
为了代码的美观，这里写死配置，都定义的常量，如果真正使用的话，建议这里修改成 读取配置方式 例如 nacos
/**
* @author caicai
*/
public class MqConfigConstant {
public class MqClientConfig {
// 用户名称
public static final String MQ_USER_NAME = "test";
// 用户密码
public static final String MQ_PASS_WORD = "test";
// 默认主题
public static final String MQ_DEFAULT_TOPIC = "test";
// 客户端ID
public static final String MQ_CLIENT_ID = "TEST_MQ";
// 连接地址
public static final String MQ_CONNECT_URL = "tcp://47.102.199.23:1883";
}
public class ConstantNumber {
// 默认消息质量
public static final int MQ_QOS1 = 1;
}
}
编写一个初始化配置类
这里本来是打算使用单例的模式进行配置的，写了一个单例的工具类，后面觉得这样方式 很麻烦 ，于是决定 修改成一次性初始化的
package com.xx.xx.mqtt.config;
import com.xx.xx.constant.MqConfigConstant;
import com.xxx.xx.mqtt.xx.PushCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
/**
* @author caicai
*/
@Component
public class MqClientConfig {
private MqttClient mqttClientInit;
@Bean
public MqttClient getMqttClient() {
// 首先判断一下是否已经初始化过，如果已经初始化过，则直接返回
if(mqttClientInit == null){
synchronized (MqClientConfig.class){
if(mqttClientInit == null){
try {
MemoryPersistence persistence = new MemoryPersistence();
MqttClient client = new MqttClient(MqConfigConstant.MqClientConfig.MQ_CONNECT_URL, MqConfigConstant.MqClientConfig.MQ_CLIENT_ID, persistence);
// 设置连接时协带参数
MqttConnectOptions connOpts = new MqttConnectOptions();
// 设置连接时候需要用的用户名和密码
connOpts.setUserName(MqConfigConstant.MqClientConfig.MQ_USER_NAME);
connOpts.setPassword(MqConfigConstant.MqClientConfig.MQ_PASS_WORD.toCharArray());
// 连接
client.connect(connOpts);
// 调用之前的回调方法类
client.setCallback(new PushCallback());
// 设置默认订阅 -这里我为了方便测试，加的。可以去掉
client.subscribe(MqConfigConstant.MqClientConfig.MQ_DEFAULT_TOPIC);
mqttClientInit = client;
}catch (Exception e){
System.out.println("初始化MqttClient失败");
}
}
}
}
return mqttClientInit;
}
}
写提供实现服务，来方便后续的调用 例如跨服务调用
@Service
public class MqttService {
@Resource
private MqttClient mqttClient;
public void setSubscribe(String subTopic) throws MqttException {
mqttClient.subscribe(subTopic);
}
public void setPublish(String pubTopic,String msg) throws MqttException {
MqttMessage message = new MqttMessage(msg.getBytes());
mqttClient.publish(pubTopic,message);
}
public void setPublish(String pubTopic,String msg,int qos) throws MqttException {
MqttMessage message = new MqttMessage(msg.getBytes());
message.setQos(qos);
mqttClient.publish(pubTopic,message);
}
}