EMQX搭建及Android端连接MQTT消息服务器

一、相关资料

1、MQTT官网：http://mqtt.org/
2、EMQX官网：https://www.emqx.io/cn/
3、EMQX用户手册：https://docs.emqx.io/docs/broker/v3/cn/guide.html
4、MQTT API：http://www.eclipse.org/paho/files/javadoc/index.html
5、MQTT Android API： http://www.eclipse.org/paho/files/android-javadoc/index.html

二、MQTT协议简介

1、MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上。
2、MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。
3、MQTT协议的三种身份：
发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。
其中MQTT消息分为：主题（Topic）和负载（Payload）两部分：

（1）Topic，可以理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（Payload）；
（2）Payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

三、EMQ X简介

1、EMQ X (Erlang/Enterprise/Elastic MQTT Broker) 是基于 Erlang/OTP 平台开发的开源物联网 MQTT 消息服务器。Erlang/OTP 是出色的软实时(Soft-Realtime)、低延时(Low-Latency)、分布式(Distributed) 的语言平台。MQTT 是轻量的(Lightweight)、发布订阅模式(PubSub) 的物联网消息协议。

四、EMQ X消息平台搭建

1、Windows

（1）下载 EMQ X Broker https://www.emqx.io/cn/downloads
（2）解压下载的ZIP
（3）打开 Windows 命令行窗口，cd 到程序目录，启动 EMQ X。

 /*我的程序在D盘*/
cd /d D:\emqx\bin
emqx start

这样就启动MQTT服务器了

EMQ X 消息服务器默认占用的 TCP 端口包括:
1883 MQTT 协议端口
8883 MQTT/SSL 端口
8083 MQTT/WebSocket 端口
8081 HTTP API 端口
18083 Dashboard 管理控制台端口

（4）进入EMQX控制台界面 http://127.0.0.1:18083/#/websocket
登陆后，用户名和密码的默认值为 admin 和 public
（5）测试
- 1、进入工具->websocket
  点击连接
- 2、监控 -> 客户端
  就可以看见上线的客户端
  注意：客户端ID必须唯一
- 3、发送信息
  进入工具->websocket
  
  先点击订阅然后点击发送

搭建完成

五、Android端搭建

1、在app中的build.gradle dependencies 下引入依赖资源

 implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.2'
 implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

2、在Manifest添加服务

    <uses-permission android:name="android.permission.INTERNET" />
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
    <uses-permission android:name="android.permission.WAKE_LOCK" />

3、在Manifest中注册Service

 <service android:name="org.eclipse.paho.android.service.MqttService" />
 <service android:name=".MqttServer"/>

4、相关类型说明
CLIENT_ID：客户身份唯一标识。
PUBLISH_TOPIC：在 MQTT 中订阅者( Subscribe )同一主题（Topic）的客户端会同时收到消息推送。
RESPONSE_TOPIC 获得响应之后的回传
HOST：服务器地址（协议+地址+端口号）
USERNAME：连接MQTT服务器的用户名。
PASSWORD：连接MQTT服务器的密码。

 public static String CLIENT_ID = "HUA_WEI_P30-Pro";//客户端唯一标识
 public static String PUBLISH_TOPIC = "topic_test";//发布者主题
 public static String RESPONSE_TOPIC = "message_arrived";//响应主题
 public String HOST = "tcp://192.168.10.101:1883";//服务器地址（协议+地址+端口号）
 public String USERNAME = "admin";
 public String PASSWORD = "public";

5、MqttAndroidClient初始化设置

    public void InitMQTTConnect() {

        mqttAndroidClient = new MqttAndroidClient(this, HOST, CLIENT_ID);

        mMqttConnectOptions = new MqttConnectOptions();
        // 在重新启动和重新连接时记住状态
        mMqttConnectOptions.setCleanSession(true);
        // 设置连接的用户名
        mMqttConnectOptions.setUserName(USERNAME);
        // 设置密码
        mMqttConnectOptions.setPassword(PASSWORD.toCharArray());
        // 设置超时时间，单位：秒
        mMqttConnectOptions.setConnectionTimeout(10);
        // 心跳包发送间隔，单位：秒
        mMqttConnectOptions.setKeepAliveInterval(20);
        //设置服务质量
        MqttMessage message = new MqttMessage("PayLoad".getBytes());
        message.setQos(1);

        mqttAndroidClient.setCallback(mqttCallback);// （需要实现接口回调）

        connectionMQTTServer();
    }

6、MQTT消息服务器连接

        private void connectionMQTTServer() { // 连接操作

            new Thread(new Runnable() {
            @Override
            public void run() {

                try {
                    //进行服务器连接
                    /***
                     * mMqttConnectOptions MQTT设置
                     * iMqttActionListener MQTT连接监听
                     */
                    mqttAndroidClient.connect(mMqttConnectOptions, null, new IMqttActionListener() {
                        @Override
                        public void onSuccess(IMqttToken asyncActionToken) { // 连接成功

                            try {
                                Toast.makeText(GlobContext.GetGlobContext(), "连接成功！", Toast.LENGTH_LONG).show();
                                mqttAndroidClient.subscribe(PUBLISH_TOPIC, 1);//订阅主题，参数：主题、服务质量
                            } catch (MqttException e) {
                                e.printStackTrace();
                            }
                        }

                        @Override
                        public void onFailure(IMqttToken asyncActionToken, Throwable exception) { //连接失败
                            Toast.makeText(GlobContext.GetGlobContext(), "连接失败！正在重新连接！", Toast.LENGTH_LONG).show();
                            exception.printStackTrace();
                            new Handler().postDelayed(new Runnable() {
                                @Override
                                public void run() {
                                    connectionMQTTServer(); // ReConnection
                                }
                            }, 5000);   //延时5秒重新连接MQTT服务器
                        }
                    });

                } catch (MqttException e) {
                    e.fillInStackTrace();
                }
            }
        }).run();

    }

7、连接之后实现mqttCallback接口

 private MqttCallback mqttCallback = new MqttCallbackExtended() {  //回传
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            /**
             *与服务器的连接成功完成时调用。
             * @param reconnect如果为true，则连接是自动重新连接的结果。
             * @param serverURI建立连接的服务器URI。
             **/

        }

        @Override
        public void connectionLost(Throwable cause) {

            Log.i(TAG, "连接断开 ");
            connectionMQTTServer(); // ReConnection
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {  // 接收的消息

            String s = new String(message.getPayload(), "GB2312");

            Log.e(TAG, topic + s);  //接收的消息
            Toast.makeText(GlobContext.GetGlobContext(), s, Toast.LENGTH_LONG).show();
            response("GetInfo");  //获得消息进行回传

        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {

        }
    };

8、回应

    public void response(String message) {
        String topic = RESPONSE_TOPIC;
        Integer qos = 1;
        Boolean retained = false;
        try {
            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
            mqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

9、订阅者Subscribe 在与服务器进行连接时进行消息订阅

//订阅主题，参数：主题、服务质量
 mqttAndroidClient.subscribe(PUBLISH_TOPIC,  qos.intValue());

10、发布者Publish

//参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
mqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());

11、完整代码

package com.example.emqttlinkdemo;

import android.app.Service;
import android.content.Context;
import android.content.Intent;
import android.net.ConnectivityManager;
import android.net.NetworkInfo;
import android.os.Handler;
import android.os.IBinder;
import android.util.Log;
import android.widget.Toast;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;


public class MqttServer extends Service {

    private static String TAG = "Android_MQTT_Demo";

    private static MqttAndroidClient mqttAndroidClient;
    private MqttConnectOptions mMqttConnectOptions;

    public String HOST = "tcp://192.168.10.101:1883";//服务器地址（协议+地址+端口号）
    public String USERNAME = "admin";//用户名
    public String PASSWORD = "public";//密码

    public static String PUBLISH_TOPIC = "topic_test";//发布主题
    public static String RESPONSE_TOPIC = "message_arrived";//响应主题

    public String CLIENT_ID = "HUA_WEI_P30-Pro";


    @Override
    public void onCreate() {
        super.onCreate();
        InitMQTTConnect();
    }

    public void InitMQTTConnect() {

        mqttAndroidClient = new MqttAndroidClient(this, HOST, CLIENT_ID);

        mMqttConnectOptions = new MqttConnectOptions();
        // 在重新启动和重新连接时记住状态
        mMqttConnectOptions.setCleanSession(true);
        // 设置连接的用户名
        mMqttConnectOptions.setUserName(USERNAME);
        // 设置密码
        mMqttConnectOptions.setPassword(PASSWORD.toCharArray());
        // 设置超时时间，单位：秒
        mMqttConnectOptions.setConnectionTimeout(10);
        // 心跳包发送间隔，单位：秒
        mMqttConnectOptions.setKeepAliveInterval(20);
        //设置服务质量
        MqttMessage message = new MqttMessage("PayLoad".getBytes());
        message.setQos(1);

        mqttAndroidClient.setCallback(mqttCallback);// 回调

        connectionMQTTServer();
    }

    /**
     * 进行连接操作
     */

    private void connectionMQTTServer() { // 连接操作

        new Thread(new Runnable() {
            @Override
            public void run() {

                try {
                    //进行服务器连接
                    /***
                     * mMqttConnectOptions MQTT设置
                     * iMqttActionListener MQTT连接监听
                     */
                    mqttAndroidClient.connect(mMqttConnectOptions, null, new IMqttActionListener() {
                        @Override
                        public void onSuccess(IMqttToken asyncActionToken) { // 连接成功

                            try {
                                Toast.makeText(GlobContext.GetGlobContext(), "连接成功！", Toast.LENGTH_LONG).show();
                                mqttAndroidClient.subscribe(PUBLISH_TOPIC, 1);//订阅主题，参数：主题、服务质量
                            } catch (MqttException e) {
                                e.printStackTrace();
                            }
                        }

                        @Override
                        public void onFailure(IMqttToken asyncActionToken, Throwable exception) { //连接失败
                            Toast.makeText(GlobContext.GetGlobContext(), "连接失败！正在重新连接！", Toast.LENGTH_LONG).show();
                            exception.printStackTrace();
                            new Handler().postDelayed(new Runnable() {
                                @Override
                                public void run() {
                                    connectionMQTTServer(); // ReConnection
                                }
                            }, 5000);   //延时5秒重新连接MQTT服务器
                        }
                    });

                } catch (MqttException e) {
                    e.fillInStackTrace();
                }
            }
        }).run();

    }


    private MqttCallback mqttCallback = new MqttCallbackExtended() {  //回传
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            /**
             *与服务器的连接成功完成时调用。
             * @param reconnect如果为true，则连接是自动重新连接的结果。
             * @param serverURI建立连接的服务器URI。
             **/

        }

        @Override
        public void connectionLost(Throwable cause) {

            Log.i(TAG, "连接断开 ");
            connectionMQTTServer(); // ReConnection
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {  // 接收的消息

            String s = new String(message.getPayload(), "GB2312");

            Log.e(TAG, topic + s);  //接收的消息
            Toast.makeText(GlobContext.GetGlobContext(), s, Toast.LENGTH_LONG).show();
            response("GetInfo");

        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {

        }
    };

    /**
     * 发布消息 publish(主题,消息的字节数组,服务质量,是否在服务器保留断开连接后的最后一条消息);
     *
     * @param message
     */

    /**
     * 将消息发布到服务器上的主题。
     * <p>
     * 一种方便的方法，它将创建一个新的{@link MqttMessage}对象
     * 具有字节数组有效负载和指定的QoS，然后将其发布。
     * </ p>
     *
     * @throws IllegalArgumentException 如果QoS的值不为0、1或2。
     * @param主题 将消息传递到例如“ finance / stock / ibm”。
     * @参数有效负载 用作有效载荷的字节数组
     * @参数qos 提供消息的服务质量。有效值是0、1或2。
     * @param保留 服务器是否应保留此消息。
     * @return令牌用于跟踪并等待发布完成。的 令牌将传递给已设置的任何回调。
     * @抛出MqttPersistenceException 发生问题时存储消息
     * @抛出MqttException 用于发布消息时遇到的其他错误。
     * 例如，正在处理太多消息。
     * @see #publish（String topic, byte[] payload, int qos,boolean retained）
     **/
    public static void publish(String message) {
        String topic = PUBLISH_TOPIC;
        Boolean retained = false;
        try {
            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
            mqttAndroidClient.publish(topic, message.getBytes(), 1, retained.booleanValue());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    public void response(String message) {
        String topic = RESPONSE_TOPIC;
        Integer qos = 1;
        Boolean retained = false;
        try {
            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
            mqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 将通信通道返回到服务。如果可能返回null
     * 客户端无法绑定到服务。返回的
     * {@link IBinder}通常用于复杂的界面
     * 已<a href="{@docRoot}guide/components/aidl.html">描述为aidl </a>。
     * <p> <em>请注意，与其他应用程序组件不同，调用
     * 此处返回的IBinder接口可能不会在主线程上发生
     * 的过程</ em>。有关主线程的更多信息，请参见
     * <a href="{@docRoot}guide/topics/fundamentals/processes-and-threads.html">流程和线程</a>。</ p>    
     *
     * @param intent 用于绑定到此服务的Intent，
     *               如{@link Context＃bindServiceContext.bindService}。请注意，随附的所有其他功能此时的意图不会在这里显示。
     * @return返回IBinder，客户端可以通过该IBinder调用 服务。
     **/
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    @Override
    public void onDestroy() {
        super.onDestroy();
        try {
            mqttAndroidClient.disconnect();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

10、测试功能

最后编辑于：2020.04.26 19:45:21

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 203,362评论 5赞 477
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 85,330评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 150,247评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,560评论 1赞 273
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,580评论 5赞 365
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,569评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 37,929评论 3赞 395
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,587评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 40,840评论 1赞 297
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,596评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,678评论 1赞 329
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,366评论 4赞 318
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 38,945评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,929评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,165评论 1赞 259
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 43,271评论 2赞 349
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,403评论 2赞 342