MQTT java 实现 案例 demo

一.MQTT介绍

1.简介

MQTT（message queuing telemetry transport）是IBM开发的即时通讯协议，是一种发布/订阅极其轻量级的消息传输协议，专门为网络受限设备、低宽带以及高延迟和不可靠的网络而设计的。由于以上轻量级的特点，是实现智能家居的首选传输协议，相比于XMPP，更加轻量级而且占用宽带低。

2.特点

a.由于采用发布/订阅的消息模式，可以提供一对多的消息发布
b.轻量级，网络开销小
c.对负载内容会有屏蔽的消息传输
d.有三种消息发布质量（Qos）：
qos=0：“至多一次”，这一级别会发生消息丢失或重复，消息发布依赖于TCP/IP网络
qos=1：“至少一次”，确保消息到达，但消息重复可能会发生
qos=2：“只有一次”，确保消息到达一次
e.通知机制，异常中断时会通知双方

3.原理

 

 

MQTT协议有三种身份：发布者、代理、订阅者，发布者和订阅者都为客户端，代理为服务器，同时消息的发布者也可以是订阅者（为了节约内存和流量发布者和订阅者一般都会定义在一起）。
MQTT传输的消息分为主题（Topic，可理解为消息的类型，订阅者订阅后，就会收到该主题的消息内容（payload））和负载（payload，可以理解为消息的内容）两部分。

二.MQTT DEMO

windos下 下载并安装mqtt 服务 运行并通过浏览器打开  http://127.0.0.1:18083/#/connections

MQTT 启动说明：
进入bin目录用cmd方式启动
启动命令emqx start，
关闭 emqx stop， 注意是 bin目录下。

管理网页http://192.168.128.8:18083/#/topics

账户 admin/public

jar包：主要是paho包

java demo 代码 

1.监控客户端状态：

package com.yt.mqtt.simple;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
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.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;

/**
 * 监控客户端状态
 * 
 * @author YT
 * 2020-05-08
 */
public class ClientStatus {
     public static void main(String[] args) throws MqttException {   
            String HOST = "tcp://127.0.0.1:11883";  
//            String TOPIC = "mqtt/test";
//            String TOPIC = "$SYS/broker/clients/connected";
            String TOPIC ="$SYS/brokers/+/clients/#";
            int qos = 1;
            String clientid = "subClient2";
            String userName = "admin";
            String passWord = "test";
            try {
                // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
                MqttClient client = new MqttClient(HOST, clientid, new MemoryPersistence());
                // MQTT的连接设置
                MqttConnectOptions 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 MqttCallback() {

                    public void connectionLost(Throwable cause) {
                        System.out.println("connectionLost");
                    }

//                    public void messageArrived(String topic, MqttMessage message) throws Exception {
//                        System.out.println("topic:"+topic);
//                        System.out.println("Qos:"+message.getQos());
//                        System.out.println("message content:"+new String(message.getPayload()));
//                        
//                    }

                    public void deliveryComplete(IMqttDeliveryToken token) {
                        System.out.println("deliveryComplete---------"+ token.isComplete());
                    }
                    
                    public void messageArrived(String topic, MqttMessage message) throws Exception {
                        String msg = new String(message.getPayload());
//                        System.out.println("topic:"+topic);
//                        System.out.println("Qos:"+message.getQos());
                        System.out.println("Message content:"+msg); 
                        try {
                            JSONObject jsonObject = JSON.parseObject(msg);
                            String clientId = String.valueOf(jsonObject.get("clientid"));
                            if (topic.endsWith("disconnected")) {
                                 System.out.println("客户端已掉线：" +clientId);
                            } else {
                                  System.out.println("客户端已上线："+clientId);
                            }
                        } catch (JSONException e) {
                              System.out.println("JSON Format Parsing Exception : "+ msg);
                        }
                    }
                });
                client.connect(options);
                //订阅消息
                client.subscribe(TOPIC, qos);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
}

消息发布类

package com.yt.mqtt.simple;

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.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 消息发布
 * 
 * @author YT
 * 2020-05-08
 */
public class PublishSample {
    public static void main(String[] args) {

        String topic = "mqtt/test";
        String content = "hello 哈哈";
        int qos = 1;
        String broker = "tcp://127.0.0.1:11883";  
        String userName = "admin";
        String password = "test";
        String clientId = "pubClient";
        // 内存存储
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            // 创建客户端
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            // 创建链接参数
            MqttConnectOptions connOpts = new MqttConnectOptions();
            // 在重新启动和重新连接时记住状态
            connOpts.setCleanSession(false);
            // 设置连接的用户名
            connOpts.setUserName(userName);
            connOpts.setPassword(password.toCharArray());
            // 建立连接
            sampleClient.connect(connOpts);
            // 创建消息
            MqttMessage message = new MqttMessage(content.getBytes());
            // 设置消息的服务质量
            message.setQos(qos);
            // 发布消息
            sampleClient.publish(topic, message);
            // 断开连接
            sampleClient.disconnect();
            // 关闭客户端
            sampleClient.close();
        } catch (MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }
}

消息订阅类

package com.yt.mqtt.simple;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
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.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


/**
 * 消息订阅
 * 
 * @param args
 * @throws MqttException
 */
public class SubscribeSample {
//    <dependencies>
//    <dependency>
//        <groupId>org.eclipse.paho</groupId>
//        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
//        <version>1.2.0</version>
//    </dependency>
//</dependencies>

    public static void main(String[] args) throws MqttException {
        String HOST = "tcp://127.0.0.1:11883";
        String TOPIC = "mqtt/test";
        int qos = 1;
        String clientid = "subClient1";
        String userName = "admin";
        String passWord = "test";
        try {
            // host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
            MqttClient client = new MqttClient(HOST, clientid, new MemoryPersistence());
            // MQTT的连接设置
            MqttConnectOptions 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 MqttCallback() {

                public void connectionLost(Throwable cause) {
                    System.out.println("connectionLost");
                }

                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("topic:" + topic);
                    System.out.println("Qos:" + message.getQos());
                    System.out.println("message content:" + new String(message.getPayload()));

                }

                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("deliveryComplete---------" + token.isComplete());
                }

            });
            client.connect(options);
            // 订阅消息
            client.subscribe(TOPIC, qos);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

执行结果：

ClientStatus：

Message content:{"clean_start":true,"clientid":"subClient1","connack":0,"ipaddress":"127.0.0.1","keepalive":20,"proto_name":"MQTT","proto_ver":4,"ts":1588933626,"username":"admin"}
客户端已上线：subClient1
Message content:{"clean_start":false,"clientid":"pubClient","connack":0,"ipaddress":"127.0.0.1","keepalive":60,"proto_name":"MQTT","proto_ver":4,"ts":1588933634,"username":"admin"}
客户端已上线：pubClient
Message content:{"clientid":"pubClient","username":"admin","reason":"normal","ts":1588933634}
客户端已掉线：pubClient

SubscribeSample：

topic:mqtt/test
Qos:1
message content:hello 哈哈