mqtt安装和使用

linux下 下载:

wget https://www.emqx.io/downloads/broker/v3.2.1/emqx-centos7-v3.2.1.zip

解压：unzip emqx-centos7-v3.2.1.zip

启动：./emqx start

配置文件修改：

/usr/local/mqtt/emqx/etc/emqx.conf

修改账号密码：

设置密码认证：

allow_anonymous = false

导入插件：

cd /usr/lib/emqx/bin
sh emqx_ctl plugins load emqx_auth_username

设置账号密码

emqx_ctl users add admin public

springboot集成:

pom：

<!--mqtt-->
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.0</version>
</dependency>
<dependency>
    <groupId>org.fusesource.mqtt-client</groupId>
    <artifactId>mqtt-client</artifactId>
    <version>1.14</version>
</dependency>

yml配置：

spring:

  mqtt:
    username: xxx
    password: xxx
    url: tcp://www.xxx.cn:1883
    client:
      id: xxx
    topic: xxx
    completionTimeout: 3000

package com.cw.common.mqtt;
/**
 * @Description:
 * @Auther: CW
 * @Date: 2021/8/5 16:40
 */
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
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.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
/**
 * MQTT客户端订阅消息类
 * @author zhongyulin
 *
 */
@Component
public class MqttConsumer implements ApplicationRunner {

    private static Logger logger = LoggerFactory.getLogger(MqttConsumer.class);

    private static MqttClient client;

    private static MqttTopic mqttTopic;

    /**
     * MQTT连接属性配置对象
     */
    @Autowired
    public MqttCofigBean mqttCofigBean;

    /**
     * 初始化参数配置
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("初始化启动MQTT连接");
        this.connect();
    }


    /**
     * 用来连接服务器
     */
    private void connect() throws Exception {
        client = new MqttClient(mqttCofigBean.getHostUrl(), mqttCofigBean.getClientId(), new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(mqttCofigBean.getUsername());
        options.setPassword(mqttCofigBean.getPassword().toCharArray());
        options.setCleanSession(false);   //是否清除session
        // 设置超时时间
        options.setConnectionTimeout(30);
        // 设置会话心跳时间
        options.setKeepAliveInterval(20);
        try {
            String[] msgtopic = mqttCofigBean.getMsgTopic();
            //订阅消息
            int[] qos = new int[msgtopic.length];
            for (int i = 0; i < msgtopic.length; i++) {
                qos[i] = 0;
            }
            client.setCallback(new TopMsgCallback(client, options, msgtopic, qos));
            client.connect(options);
            client.subscribe(msgtopic, qos);
            logger.info("MQTT连接成功:" + mqttCofigBean.getClientId() + ":" + client);
        } catch (Exception e) {
            logger.error("MQTT连接异常：" + e);
        }
    }


    /**
     * 重连
     *
     * @throws Exception
     */
    public void reConnect() throws Exception {
        if (null != client) {
            this.connect();
        }
    }

    /**
     * 订阅某个主题
     *
     * @param topic
     * @param qos
     */
    public void subscribe(String topic, int qos) {
        try {
            logger.info("topic:" + topic);
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public MqttClient getClient() {
        return client;
    }

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

    public MqttTopic getMqttTopic() {
        return mqttTopic;
    }

    public void setMqttTopic(MqttTopic mqttTopic) {
        this.mqttTopic = mqttTopic;
    }
}

package com.cw.common.mqtt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Component
public class MqttCofigBean {

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.url}")
    private String hostUrl;

    @Value("${spring.mqtt.client.id}")
    private String clientId;

    @Value("${spring.mqtt.topic}")
    private String msgTopic;

    @Value("${spring.mqtt.completionTimeout}")
    private int completionTimeout;   //连接超时
    /**
     * 获取用户名
     *
     * @return
     */
    public String getUsername() {
        return this.username;
    }

    /**
     * 获取密码
     *
     * @return
     */
    public String getPassword() {
        return this.password;
    }

    /**
     * 获取服务器连接地址
     *
     * @return
     */
    public String getHostUrl() {
        return this.hostUrl;
    }

    /**
     * 获取客户端ID
     *
     * @return
     */
    public String getClientId() {
        return this.clientId;
    }

    /**
     * 获取默认主题
     *
     * @return
     */
    public String[] getMsgTopic() {
        String[] topic = msgTopic.split(",");
        return topic;
    }

    /***
     * 获取连接超时时间
     * @return
     */
    public int getCompletionTimeout() {
        return this.completionTimeout;
    }
}

package com.cw.common.mqtt;

/**
 * @Description:
 * @Auther: CW
 * @Date: 2021/8/5 16:41
 */

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

/**
 * MQTT消息处理类
 * @author zhongyulin
 *  */
public class TopMsgCallback implements MqttCallback {

    private static Logger logger = LoggerFactory.getLogger(TopMsgCallback.class);

    private MqttClient client;
    private MqttConnectOptions options;
    private String[] topic;
    private int[] qos;

    public TopMsgCallback() {
    }

    public TopMsgCallback(MqttClient client, MqttConnectOptions options, String[] topic, int[] qos) {
        this.client = client;
        this.options = options;
        this.topic = topic;
        this.qos = qos;
    }

    /**
     * 断开重连
     */
    public void connectionLost(Throwable cause) {
        logger.info("MQTT连接断开，发起重连");
        while (true) {
            try {
                Thread.sleep(30000);
                client.connect(options);
                //订阅消息
                client.subscribe(topic, qos);
                logger.info("MQTT重新连接成功:" + client);
                break;
            } catch (Exception e) {
                e.printStackTrace();
                continue;
            }
        }

    }

    /**
     * 接收到消息调用令牌中调用
     */
    public void deliveryComplete(IMqttDeliveryToken token) {
    }

    /**
     * 消息处理
     */
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println();
        //订阅消息字符
        String msg = new String(message.getPayload());
//        byte[] bymsg = getBytesFromObject(msg);
        logger.info("topic:" + topic);
        logger.info("msg:" + msg);
    }

    //对象转化为字节码
//    public byte[] getBytesFromObject(Serializable obj) throws Exception {
//        if (obj == null) {
//            return null;
//        }
//        ByteArrayOutputStream bo = new ByteArrayOutputStream();
//        ObjectOutputStream oo = new ObjectOutputStream(bo);
//        oo.writeObject(obj);
//        return bo.toByteArray();
//    }
}