springboot集成mqtt(超级无敌详细)

在这里插入图片描述

springboot集成MQTT步骤

1. 引入pom依赖

<!-- mqtt --><dependency><groupId>org.springframework.integration</groupId><artifactId>spring-integration-mqtt</artifactId></dependency>

2. application.yml

## MQTT##
mqtt:
    host: tcp://192.168.10.198:1883
    userName: root
    passWord: 123456
    qos: 1
    clientId: ClientId_local #ClientId_local必须唯一 比如你已经定了叫ABC  那你就一直叫ABC  其他地方就不要使用ABC了
    timeout: 10
    keepalive: 20
    topic1: A/pick/warn/#  #符号是代表整个warn下面的全部子主题 没有理解的话 可以百度仔细理解一下
    topic2: A/cmd/resp
    topic3: ABCF
    topic4: ABCH

application.properties

## MQTT##mqtt.host=tcp://192.168.10.198:1883
mqtt.clientId=ClientId_local
mqtt.username=admin
mqtt.password=123456mqtt.timeout=10mqtt.keepalive=20mqtt.topic1=A/pick/warn/#

3. MqttConfiguration.java

import org.eclipse.paho.client.mqttv3.MqttException;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;

/**
 1. @author WXY
 2. @date 2022/6/29 20:42
 */
@Configuration
public class MqttConfiguration {

    private static final Logger log = LoggerFactory.getLogger(MqttConfiguration.class);
    @Value("${mqtt.host}")
    String host;
    @Value("${mqtt.username}")
    String username;
    @Value("${mqtt.password}")
    String password;
    @Value("${mqtt.clientId}")
    String clientId;
    @Value("${mqtt.timeout}")
    int timeOut;
    @Value("${mqtt.keepalive}")
    int keepAlive;
    @Value("${mqtt.topic1}")
    String topic1;
    @Value("${mqtt.topic2}")
    String topic2;
    @Value("${mqtt.topic3}")
    String topic3;
    @Value("${mqtt.topic4}")
    String topic4;

    @Bean//注入spring
    public MyMQTTClient myMQTTClient(){
        MyMQTTClient myMQTTClient = new MyMQTTClient(host, username, password, clientId, timeOut, keepAlive);for(int i =0; i <10; i++){
            try {
                myMQTTClient.connect();
                //不同的主题
             //   myMQTTClient.subscribe(topic1, 1);
             //   myMQTTClient.subscribe(topic2, 1);
             //   myMQTTClient.subscribe(topic3, 1);
             //   myMQTTClient.subscribe(topic4, 1);return myMQTTClient;} catch (MqttException e){
                log.error("MQTT connect exception,connect time = " + i);
                try {
                    Thread.sleep(2000);} catch (InterruptedException e1){
                    e1.printStackTrace();}}}return myMQTTClient;}
     public String getTopic1(){return topic1;}

    public void setTopic1(String topic1){
        this.topic1 = topic1;}

    public String getTopic2(){return topic2;}

    public void setTopic2(String topic2){
        this.topic2 = topic2;}

    public String getTopic3(){return topic3;}

    public void setTopic3(String topic3){
        this.topic3 = topic3;}

    public String getTopic4(){return topic4;}

    public void setTopic4(String topic4){
        this.topic4 = topic4;}}

4. MyMQTTClient.java

import org.eclipse.paho.client.mqttv3.*;import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;import org.slf4j.Logger;import org.slf4j.LoggerFactory;

/**
 1. @author WXY
 2. @date 2022/6/29 20:43
 */
public class MyMQTTClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyMQTTClient.class);

    private static MqttClient client;
    private String host;
    private String username;
    private String password;
    private String clientId;
    private int timeout;
    private int keepalive;
    public MyMQTTClient(String host, String username, String password, String clientId, int timeOut, int keepAlive){
        this.host =host;
        this.username = username;
        this.password = password;
        this.clientId = clientId;
        this.timeout = timeOut;
        this.keepalive = keepAlive;}

    public static MqttClient getClient(){return client;}

    public static void setClient(MqttClient client){
        MyMQTTClient.client = client;}

    /**
     * 设置mqtt连接参数
     *
     * @param username
     * @param password
     * @param timeout
     * @param keepalive
     * @return
     */
    public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive){
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(timeout);
        options.setKeepAliveInterval(keepalive);
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);return options;}

    /**
     * 连接mqtt服务端，得到MqttClient连接对象
     */
    public void connect() throws MqttException {if(client == null){
            client = new MqttClient(host, clientId, new MemoryPersistence());
            client.setCallback(new MyMQTTCallback(MyMQTTClient.this));}
        MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);if(!client.isConnected()){
            client.connect(mqttConnectOptions);}else{
            client.disconnect();
            client.connect(mqttConnectOptions);}
        LOGGER.info("MQTT connect success");//未发生异常，则连接成功
    }

    /**
     * 发布，默认qos为0，非持久化
     *
     * @param pushMessage
     * @param topic
     */
    public void publish(String pushMessage, String topic){
        publish(pushMessage, topic, 0, false);}

    /**
     * 发布消息
     *
     * @param pushMessage
     * @param topic
     * @param qos
     * @param retained:留存
     */
    public void publish(String pushMessage, String topic, int qos, boolean retained){
        MqttMessage message = new MqttMessage();
        message.setPayload(pushMessage.getBytes());
        message.setQos(qos);
        message.setRetained(retained);
        MqttTopic mqttTopic = MyMQTTClient.getClient().getTopic(topic);if(null == mqttTopic){
            LOGGER.error("topic is not exist");}
        MqttDeliveryToken token;//Delivery:配送
        synchronized (this){//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
            try {
                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
                token.waitForCompletion(1000L);} catch (MqttPersistenceException e){
                e.printStackTrace();} catch (MqttException e){
                e.printStackTrace();}}}

    /**
     * 订阅某个主题
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos){
        try {
            MyMQTTClient.getClient().subscribe(topic, qos);} catch (MqttException e){
            e.printStackTrace();}}

    /**
     * 取消订阅主题
     *
     * @param topic 主题名称
     */
    public void cleanTopic(String topic){if(client != null && client.isConnected()){
            try {
                client.unsubscribe(topic);} catch (MqttException e){
                e.printStackTrace();}}else{
            System.out.println("取消订阅失败！");}}}

5. MyMQTTCallback.java

import cn.hutool.core.util.CharsetUtil;import com.alibaba.fastjson.JSON;import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;import org.eclipse.paho.client.mqttv3.MqttCallback;import org.eclipse.paho.client.mqttv3.MqttException;import org.eclipse.paho.client.mqttv3.MqttMessage;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.Map;

/**
 * @author WXY
 * @date 2022/6/29 20:43
 */
public class MyMQTTCallback implements MqttCallbackExtended {

    //手动注入
    private MqttConfiguration mqttConfiguration = SpringUtils.getBean(MqttConfiguration.class);

    private static final Logger log = LoggerFactory.getLogger(MyMQTTCallback.class);

    private MyMQTTClient myMQTTClient;

    public MyMQTTCallback(MyMQTTClient myMQTTClient){
        this.myMQTTClient = myMQTTClient;}

   /**
     * 丢失连接，可在这里做重连
     * 只会调用一次
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable){
        log.error("mqtt connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());
        long reconnectTimes =1;while(true){
            try {if(MyMQTTClient.getClient().isConnected()){
                    //判断已经重新连接成功  需要重新订阅主题 可以在这个if里面订阅主题  或者 connectComplete（方法里面）  看你们自己选择
                    log.warn("mqtt reconnect success end  重新连接  重新订阅成功");return;}reconnectTimes+=1;
                log.warn("mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
                MyMQTTClient.getClient().reconnect();} catch (MqttException e){
                log.error("mqtt断连异常", e);}
            try {
                Thread.sleep(5000);} catch (InterruptedException e1){}}}

    /**
     * @param topic
     * @param mqttMessage
     * @throws Exception
     * subscribe后得到的消息会执行到这里面
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        log.info("接收消息主题 : {}，接收消息内容 : {}", topic, new String(mqttMessage.getPayload()));
        //发布消息主题
        if(topic.equals("embed/resp")){
            Map maps =(Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
            //你自己的业务接口
            insertCmdResults(maps);}
        //接收报警主题
        if(topic.equals("embed/warn")){
            Map maps =(Map) JSON.parse(new String(mqttMessage.getPayload(), CharsetUtil.UTF_8));
            //你自己的业务接口
            insertPushAlarm(maps);}}

     /**
     *连接成功后的回调 可以在这个方法执行 订阅主题  生成Bean的 MqttConfiguration方法中订阅主题 出现bug 
     *重新连接后  主题也需要再次订阅  将重新订阅主题放在连接成功后的回调 比较合理
     * @param reconnect
     * @param serverURI
     */
    @Override
    public  void  connectComplete(boolean reconnect,String serverURI){
        log.info("MQTT 连接成功，连接方式：{}",reconnect?"重连":"直连");
        //订阅主题
        myMQTTClient.subscribe(mqttConfiguration.topic1, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic2, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic3, 1);
        myMQTTClient.subscribe(mqttConfiguration.topic4, 1);}

    /**
     * 消息到达后
     * subscribe后，执行的回调函数
     *
     * @param s
     * @param mqttMessage
     * @throws Exception
     */
    /**
     * publish后，配送完成后回调的方法
     *
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken){
        log.info("==========deliveryComplete={}==========", iMqttDeliveryToken.isComplete());}}

6. MqttMsg.java

/**
 * @author WXY
 * @date 2022/6/29 20:44
 */
public class MqttMsg {
    private String name ="";
    private String content ="";
    private String time="";

    public String getName(){return name;}

    public void setName(String name){
        this.name = name;}

    public String getContent(){return content;}

    public void setContent(String content){
        this.content = content;}

    public String getTime(){returntime;}

    public void setTime(String time){
        this.time =time;}

    @Override
    public String toString(){return"MqttMsg{" +
                "name='" + name + '\'' +
                ", content='" + content + '\'' +
                ", time='" + time + '\'' +
                '}';}}

7. MqttController.java

import com.gjwl.common.core.mqtt.MqttMsg;import com.gjwl.common.core.mqtt.MyMQTTClient;import net.sf.json.JSONObject;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Controller;import org.springframework.web.bind.annotation.PostMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.ResponseBody;import org.springframework.web.bind.annotation.RestController;import java.text.SimpleDateFormat;import java.util.Date;import java.util.LinkedList;import java.util.Queue;

/**
 * @author WXY
 * @date 2022/6/29 20:44
 */
@RestController
@RequestMapping("/sun/mqtt")
public class MqttController {

    @Autowired
    private MyMQTTClient myMQTTClient;

    @Value("${mqtt.topic1}")
    String topic1;
    @Value("${mqtt.topic2}")
    String topic2;
    @Value("${mqtt.topic3}")
    String topic3;
    @Value("${mqtt.topic4}")
    String topic4;

    Queue<String> msgQueue = new LinkedList<>();
    int count =1;

    @PostMapping("/getMsg")
    @ResponseBody
    public synchronized void mqttMsg(MqttMsg mqttMsg){
        System.out.println("队列元素数量：" + msgQueue.size());
        System.out.println("***************" + mqttMsg.getName() + ":" + mqttMsg.getContent() + "****************");

        //时间格式化
        SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String time= df.format(new Date());
        mqttMsg.setTime(time);

        mqttMsg.setContent(mqttMsg.getContent() + "——后台编号：" + count);
        count++;

        //map转json
        JSONObject json = JSONObject.fromObject(mqttMsg);
        String sendMsg = json.toString();
        System.out.println(sendMsg);

        //队列添加元素
        boolean flag = msgQueue.offer(sendMsg);if(flag){
            //发布消息  topic2 是你要发送到那个通道里面的主题 比如我要发送到topic2主题消息
            myMQTTClient.publish(msgQueue.poll(), topic2);
            System.out.println("时间戳" + System.currentTimeMillis());}
        System.out.println("队列元素数量：" + msgQueue.size());}}

8.SpringUtils.java

import org.springframework.aop.framework.AopContext;import org.springframework.beans.BeansException;import org.springframework.beans.factory.NoSuchBeanDefinitionException;import org.springframework.beans.factory.config.BeanFactoryPostProcessor;import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;import org.springframework.stereotype.Component;import com.gjwl.common.utils.StringUtils;

/**
 * spring工具类 方便在非spring管理环境中获取bean
 * 
 * @author wxy
 */
@Component
public final class SpringUtils implements BeanFactoryPostProcessor, ApplicationContextAware 
{
    /** Spring应用上下文环境 */
    private static ConfigurableListableBeanFactory beanFactory;

    private static ApplicationContext applicationContext;

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException 
    {
        SpringUtils.beanFactory = beanFactory;}

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException 
    {
        SpringUtils.applicationContext = applicationContext;}

    /**
     * 获取对象
     *
     * @param name
     * @return Object 一个以所给名字注册的bean的实例
     * @throws org.springframework.beans.BeansException
     *
     */
    @SuppressWarnings("unchecked")
    public static <T> T getBean(String name) throws BeansException
    {return(T) beanFactory.getBean(name);}

    /**
     * 获取类型为requiredType的对象
     *
     * @param clz
     * @return
     * @throws org.springframework.beans.BeansException
     *
     */
    public static <T> T getBean(Class<T> clz) throws BeansException
    {
        T result =(T) beanFactory.getBean(clz);return result;}

    /**
     * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
     *
     * @param name
     * @return boolean
     */
    public static boolean containsBean(String name){return beanFactory.containsBean(name);}

    /**
     * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
     *
     * @param name
     * @return boolean
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static boolean isSingleton(String name) throws NoSuchBeanDefinitionException
    {return beanFactory.isSingleton(name);}

    /**
     * @param name
     * @return Class 注册对象的类型
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static Class<?> getType(String name) throws NoSuchBeanDefinitionException
    {return beanFactory.getType(name);}

    /**
     * 如果给定的bean名字在bean定义中有别名，则返回这些别名
     *
     * @param name
     * @return
     * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
     *
     */
    public static String[] getAliases(String name) throws NoSuchBeanDefinitionException
    {return beanFactory.getAliases(name);}

    /**
     * 获取aop代理对象
     * 
     * @param invoker
     * @return
     */
    @SuppressWarnings("unchecked")
    public static <T> T getAopProxy(T invoker){return(T) AopContext.currentProxy();}

    /**
     * 获取当前的环境配置，无配置返回null
     *
     * @return 当前的环境配置
     */
    public static String[]getActiveProfiles(){return applicationContext.getEnvironment().getActiveProfiles();}

    /**
     * 获取当前的环境配置，当有多个环境配置时，只获取第一个
     *
     * @return 当前的环境配置
     */
    public static String getActiveProfile(){
        final String[] activeProfiles = getActiveProfiles();return StringUtils.isNotEmpty(activeProfiles) ? activeProfiles[0]: null;}}

8.测试
在这里插入图片描述
发送和接收 springboot后台日志

标签： spring boot java MQTT

本文转载自: https://blog.csdn.net/qq_42862247/article/details/125536672
版权归原作者 想要一百块 所有，如有侵权，请联系我们删除。

springboot集成mqtt(超级无敌详细)

springboot集成MQTT步骤

发表评论

“springboot集成mqtt(超级无敌详细)”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航