Ubuntu 16.04安装测试MQTT Mosquitto以及Java实例

怼烎@ 2022-04-10 14:23 755阅读 0赞

一. MQTT  
MQTT 是一个轻量级即时通信协议，使用基于 TCP/IP 协议的发布/订阅消息转发模式， MQTT 协议的中心是 broker（ 服务器/代理），客户端通过订阅消息和发布消息进行数据交互。   
若初次接触MQTT协议，可先理解以下概念：   
【MQTT协议特点】——相比于RESTful架构的物联网系统，MQTT协议借助消息推送功能，可以更好地实现远程控制。   
【MQTT协议角色】——在RESTful架构的物联网系统，包含两个角色客户端和服务器端，而在MQTT协议中包括发布者，代理器（服务器）和订阅者。   
【MQTT协议消息】——MQTT中的消息可理解为发布者和订阅者交换的内容（负载），这些消息包含具体的内容，可以被订阅者使用。   
【MQTT协议主题】——MQTT中的主题可理解为相同类型或相似类型的消息集合。

Mosquitto是一款实现了 MQTT v3.1 协议的开源消息代理软件，包括了服务器端和客户端，可以跨平台部署，提供轻量级的，支持发布/订阅的的消息推送模式，使设备对设备之间的短消息通信简单易用。

二.安装测试mosquitto  
2.1安装mosquitto

1.引入mosquitto仓库并更新

sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppa

sudo apt-get update

2.执行以下命令安装mosquitto

sudo apt-get install mosquitto

3.查看mosquitto服务状态

sudo service mosquitto status

4.开启/停止mosquitto服务

sudo service mosquitto start

sudo service mosquitto stop

2.2简单的测试

一个完整的MQTT示例包括代理器，发布者和订阅者。   
测试分为以下几个步骤：   
（1）启动服务mosquitto   
（2）订阅者通过mosquitto\_sub订阅指定主题的消息。   
（3）发布者通过mosquitto\_pub发布指定主题的消息。   
（4）代理服务器把该主题的消息推送到订阅者。

![SouthEast][]

本机测试：   
代理器，发布者和订阅者都在本机。

1.安装mosquitto客户端  
sudo apt-get install mosquitto-clients

2.打开一个终端，订阅主题

mosquitto\_sub -h localhost -t "mqtt" -v

【-h】指定要连接的MQTT服务器   
【-t】订阅主题，此处为mqtt   
【-v】打印更多的调试信息

3.打开另一个终端，发布主题

mosquitto\_pub -h localhost -t "mqtt" -m "Hello2 MQTT"

【-h】指定要连接的MQTT服务器   
【-t】向指定主题推送消息   
【-m】指定消息内容

测试结果如下，可以看到消息被显示在前一个终端上了

![è¿éåå¾çæè¿°][SouthEast 1]

mosquitto\_sub -h test.mosquitto.org -t "\#" -v

你也可以发布一些有趣的消息，所有订阅的人也同样会收到

2.也可以通过以下命令订阅特定主题，用来接收自己的消息

mosquitto\_sub -h test.mosquitto.org -t "msg\_only\_from\_me" -v

3.在另外一个终端上发布消息到这个主题，消息”My cat is Luna”便会显示在上一个终端上

mosquitto\_pub -h test.mosquitto.org -t "msg\_only\_from\_me" -m "My cat is Luna"

# 五、MQTT权限配置 #

前面我们基于Mosquitto服务器已经搭建成功了，但是默认是允许匿名用户登录，对于正式上线的项目则是需要进行用户认证（当然，用户一般都会与数据库映射，不过在这里我们就会直接将用户写入配置文件中）

1、Mosquitto服务器的配置文件为/etc/mosquitto/mosquitto.conf，关于用户认证的方式和读取的配置都在这个文件中进行

配置文件参数说明：

<table> 
 <tbody> 
  <tr> 
   <td>ID</td> 
   <td>allow_anonymous</td> 
   <td>password_file　</td> 
   <td>acl_file</td> 
   <td>result</td> 
  </tr> 
  <tr> 
   <td>1</td> 
   <td>True(默认)</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>允许匿名方式登录</td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td>False</td> 
   <td>password_file</td> 
   <td>&nbsp;</td> 
   <td>开启用户验证机制</td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td>False</td> 
   <td>password_file</td> 
   <td>acl_file</td> 
   <td>开启用户验证机制，但访问控制不起作用</td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td>True</td> 
   <td>password_file</td> 
   <td>acl_file</td> 
   <td>用户名及密码不为空，将自动进行用户验证且受到访问控制的限制；用户名及密码为空，将不进行用户验证且受到访问控制的限制</td> 
  </tr> 
  <tr> 
   <td>5</td> 
   <td>False</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>无法启动服务</td> 
  </tr> 
 </tbody> 
</table>

allow\_anonymous允许匿名

password-file密码文件

acl\_file访问控制列表

2、修改配置文件

命令：sudo vi /etc/mosquitto/mosquitto.conf

pid\_file /var/run/mosquitto.pid

persistence true  
persistence\_location /var/lib/mosquitto/

log\_dest file /var/log/mosquitto/mosquitto.log

include\_dir /etc/mosquitto/conf.d

allow\_anonymous false

password\_file /etc/mosquitto/pwdfile

acl\_file /etc/mosquitto/acl

3、添加用户信息

sudo mosquitto\_passwd -c /etc/mosquitto/pwdfile stonegeek

![1302135-20180605151704715-960809801.png][]

命令解释： -c 创建一个用户、/etc/mosquitto/pwfile.example 是将用户创建到 pwfile.example  文件中、admin 是用户名。

同样连续会提示连续输入两次密码。注意第二次创建用户时不用加 -c 如果加 -c 会把第一次创建的用户覆盖。

至此两个用户创建成功，此时如果查看 pwfile.example 文件会发现其中多了两个用户。

4、添加Topic和用户的关系

sudo vim /etc/mosquitto/acl

添加以下内容:

user stonegeek  
topic write mtopic/\#

user stonegeek  
topic read mtopic/\#

5、用户认证测试

（1）重启Mosquitto步骤

查看mosquitto的进程

命令：ps -aux|grep mosquitto

（2）杀死进程

命令：sudo kill -9 pid

（3）启动

启动服务命令：mosquitto -c /etc/mosquitto/mosquitto.conf

（4）订阅端启动（不加用户）

![1302135-20180605153026425-2133395564.png][]

订阅端启动（加用户）

mosquitto\_sub -h 192.168.2.113 -t mtopic -u stonegeek -P 123456

（5）发布端启动

mosquitto\_pub -h 192.168.2.113 -t mtopic -u stonegeek -P 123456 -m "love you"

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pIQ19iaW5nZQ_size_16_color_FFFFFF_t_70][]

# 六、MQTT实现（Java语言） #

**注意：由于我们在上面配置了MQTT的用户权限控制，所以下面的用户只能使用stonegeek登录，否则项目会运行报错，而且我们在上面设置的访问控制列表中只有mtopic主题，所以我们必须使用此主题，否则，订阅者会收不到已发布的主题内容（已经测试过了）**

下面是我们Java语言实现的MQTT服务的发布/订阅

1、添加Maven依赖

<dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.0</version>
    </dependency>

2、ServerMQTT.java

package net.xmitech.springbootmqtt;
    
    import org.eclipse.paho.client.mqttv3.MqttClient;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttException;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
    import org.eclipse.paho.client.mqttv3.MqttTopic;
    import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    /**
     * Created by StoneGeek on 2018/6/5.
     * 博客地址：http://www.cnblogs.com/sxkgeek
     * 服务器向多个客户端推送主题，即不同客户端可向服务端订阅相同的主题
     */
    public class ServerMQTT {
        //tcp://MQTT安装的服务器地址:MQTT定义的端口号
        public static final String HOST = "tcp://192.168.2.113:1883";
        //定义一个主题
        public static final String TOPIC = "mtopic";
        //定义MQTT的ID，可以在MQTT服务配置中指定
        private static final String clientid = "server11";
    
        private MqttClient client;
        private MqttTopic topic11;
        private String userName = "stonegeek";
        private String passWord = "123456";
    
        private MqttMessage message;
    
        /**
         * 构造函数
         * @throws MqttException
         */
        public ServerMQTT() throws MqttException {
            // MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, clientid, new MemoryPersistence());
            connect();
        }
    
        /**
         *  用来连接服务器
         */
        private void connect() {
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setUserName(userName);
            options.setPassword(passWord.toCharArray());
            // 设置超时时间
            options.setConnectionTimeout(10);
            // 设置会话心跳时间
            options.setKeepAliveInterval(20);
            try {
                client.setCallback(new PushCallback());
                client.connect(options);
    
                topic11 = client.getTopic(TOPIC);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         *
         * @param topic
         * @param message
         * @throws MqttPersistenceException
         * @throws MqttException
         */
        public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException,
                MqttException {
            MqttDeliveryToken token = topic.publish(message);
            token.waitForCompletion();
            System.out.println("message is published completely! "
                    + token.isComplete());
        }
    
        /**
         *  启动入口
         * @param args
         * @throws MqttException
         */
        public static void main(String[] args) throws MqttException {
            ServerMQTT server = new ServerMQTT();
    
            server.message = new MqttMessage();
            server.message.setQos(1);
            server.message.setRetained(true);
            server.message.setPayload("hello,topic11".getBytes());
            server.publish(server.topic11 , server.message);
            System.out.println(server.message.isRetained() + "------ratained状态");
        }
    }

3、ClientMQTT.java

package net.xmitech.springbootmqtt;
    import java.util.concurrent.ScheduledExecutorService;
    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;
    /**
     * Created by StoneGeek on 2018/6/5.
     * 博客地址：http://www.cnblogs.com/sxkgeek
     */
    public class ClientMQTT {
        public static final String HOST = "tcp://192.168.2.113:1883";
        public static final String TOPIC = "mtopic";
        private static final String clientid = "client11";
        private MqttClient client;
        private MqttConnectOptions options;
        private String userName = "stonegeek";
        private String passWord = "123456";
    
        private ScheduledExecutorService scheduler;
    
        private void start() {
            try {
                // host为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
                client = new MqttClient(HOST, clientid, new MemoryPersistence());
                // MQTT的连接设置
                options = new MqttConnectOptions();
                // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
                options.setCleanSession(true);
                // 设置连接的用户名
                options.setUserName(userName);
                // 设置连接的密码
                options.setPassword(passWord.toCharArray());
                // 设置超时时间 单位为秒
                options.setConnectionTimeout(10);
                // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
                options.setKeepAliveInterval(20);
                // 设置回调
                client.setCallback(new PushCallback());
                MqttTopic topic = client.getTopic(TOPIC);
                //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
                options.setWill(topic, "close".getBytes(), 2, true);
    
                client.connect(options);
                //订阅消息
                int[] Qos  = {1};
                String[] topic1 = {TOPIC};
                client.subscribe(topic1, Qos);
    
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws MqttException {
            ClientMQTT client = new ClientMQTT();
            client.start();
        }
    }

4、PushCallback.java

package net.xmitech.springbootmqtt;
    
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    
    /**
     * Created by StoneGeek on 2018/6/5.
     * 博客地址：http://www.cnblogs.com/sxkgeek
     * 发布消息的回调类
     *
     * 必须实现MqttCallback的接口并实现对应的相关接口方法CallBack 类将实现 MqttCallBack。
     * 每个客户机标识都需要一个回调实例。在此示例中，构造函数传递客户机标识以另存为实例数据。
     * 在回调中，将它用来标识已经启动了该回调的哪个实例。
     * 必须在回调类中实现三个方法：
     *
     *  public void messageArrived(MqttTopic topic, MqttMessage message)接收已经预订的发布。
     *
     *  public void connectionLost(Throwable cause)在断开连接时调用。
     *
     *  public void deliveryComplete(MqttDeliveryToken token))
     *  接收到已经发布的 QoS 1 或 QoS 2 消息的传递令牌时调用。
     *  由 MqttClient.connect 激活此回调。
     */
    public class PushCallback implements MqttCallback{
        public void connectionLost(Throwable cause) {
            // 连接丢失后，一般在这里面进行重连
            System.out.println("连接断开，可以做重连");
        }
    
        public void deliveryComplete(IMqttDeliveryToken token) {
            System.out.println("deliveryComplete---------" + token.isComplete());
        }
    
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            // subscribe后得到的消息会执行到这里面
            System.out.println("接收消息主题 : " + topic);
            System.out.println("接收消息Qos : " + message.getQos());
            System.out.println("接收消息内容 : " + new String(message.getPayload()));
        }
    }

5、结果展示

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pIQ19iaW5nZQ_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pIQ19iaW5nZQ_size_16_color_FFFFFF_t_70 2][]

致谢:

感谢这两位博主的帮助:

[https://blog.csdn.net/jsjwr/article/details/78800202][https_blog.csdn.net_jsjwr_article_details_78800202]

[https://www.cnblogs.com/sxkgeek/p/9140180.html][https_www.cnblogs.com_sxkgeek_p_9140180.html]

[SouthEast]: /images/20220328/bb4ad6e952a347d5abd7ccea50d6ff17.png
[SouthEast 1]: /images/20220328/09b070feb8ad4567a197eba8f6efd815.png
[1302135-20180605151704715-960809801.png]: /images/20220328/baaea1232a0c46beb272bc8480ffc015.png
[1302135-20180605153026425-2133395564.png]: /images/20220328/25a27766d2604304a445c368488079b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pIQ19iaW5nZQ_size_16_color_FFFFFF_t_70]: /images/20220328/ccd7b8150d6d4ac3ac85f7c1bcb777d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pIQ19iaW5nZQ_size_16_color_FFFFFF_t_70 1]: /images/20220328/c56e4e96da0a4226ab39a46d16f331d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pIQ19iaW5nZQ_size_16_color_FFFFFF_t_70 2]: /images/20220328/cf6834192a1844768c04d68783d9f48f.png
[https_blog.csdn.net_jsjwr_article_details_78800202]: https://blog.csdn.net/jsjwr/article/details/78800202
[https_www.cnblogs.com_sxkgeek_p_9140180.html]: https://www.cnblogs.com/sxkgeek/p/9140180.html