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

2020-02-17 本文已影响0人隗豪

一、相关资料

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、测试功能