Android中MQTT的使用

Android中使用MQTT需要使用到Paho Android Service库，Paho Android Service是一个用Java编写的MQTT客户端库。
GitHub地址：https://github.com/eclipse/paho.mqtt.android

1 集成

在module的build.gradle文件中添加依赖

repositories {    maven {        url "https://repo.eclipse.org/content/repositories/paho-snapshots/"    }}dependencies {    compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'    compile 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'}

在 AndroidManifest.xml 添加限权

在 AndroidManifest.xml 注册Service （MyMqttService为自己写的服务，下文会讲到）

2 具体代码

2.1 Android中使用MQTT最主要的就是以下几个方法：

connect：连接MQTT服务器，这里主要讲3个参数的方法，如下：

 @Override public IMqttToken connect(MqttConnectOptions options, Object userContext,   IMqttActionListener callback) throws MqttException {    //...}

参数options：用来携带连接服务器的一系列参数，例如用户名、密码等。
参数userContext：可选对象，用于向回调传递上下文。一般传null即可。
参数callback：用来监听MQTT是否连接成功的回调

publish：发布消息，这里使用四个参数的方法，如下：

 @Override public IMqttDeliveryToken publish(String topic, byte[] payload, int qos,   boolean retained) throws MqttException, MqttPersistenceException {    //... }

参数topic：发布消息的主题
参数payload：消息的字节数组
参数qos：提供消息的服务质量，可传0、1或2
参数retained：是否在服务器保留断开连接后的最后一条消息

subscribe：订阅消息，这里主要讲2个参数的方法，如下：

 @Override public IMqttToken subscribe(String topic, int qos) throws MqttException,   MqttSecurityException {    //... }

参数topic：订阅消息的主题
参数qos：订阅消息的服务质量，可传0、1或2

2.2 MQTT服务——MyMqttService

下面写一个 Service 来实现MQTT在Android运用中的connect、publish、subscribe

package com.wildma.mqttandroidclient;import android.app.Service;import android.content.Context;import android.content.Intent;import android.net.ConnectivityManager;import android.net.NetworkInfo;import android.os.Build;import android.os.Handler;import android.os.IBinder;import android.support.annotation.Nullable;import android.support.annotation.RequiresApi;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.MqttConnectOptions;import org.eclipse.paho.client.mqttv3.MqttException;import org.eclipse.paho.client.mqttv3.MqttMessage;/** * Author       wildma * Github       https://github.com/wildma * CreateDate   2018/11/08 * Desc         ${MQTT服务} */public class MyMqttService extends Service {    public final String TAG = MyMqttService.class.getSimpleName();    private static MqttAndroidClient  mqttAndroidClient;    private        MqttConnectOptions mMqttConnectOptions;    public        String HOST           = "tcp://192.168.0.102:61613";//服务器地址（协议+地址+端口号）    public        String USERNAME       = "admin";//用户名    public        String PASSWORD       = "password";//密码    public static String PUBLISH_TOPIC  = "tourist_enter";//发布主题    public static String RESPONSE_TOPIC = "message_arrived";//响应主题    @RequiresApi(api = 26)    public        String CLIENTID       = Build.VERSION.SDK_INT >= Build.VERSION_CODES.O            ? Build.getSerial() : Build.SERIAL;//客户端ID，一般以客户端唯一标识符表示，这里用设备序列号表示    @Override    public int onStartCommand(Intent intent, int flags, int startId) {        init();        return super.onStartCommand(intent, flags, startId);    }    @Nullable    @Override    public IBinder onBind(Intent intent) {        return null;    }    /**     * 开启服务     */    public static void startService(Context mContext) {        mContext.startService(new Intent(mContext, MyMqttService.class));    }    /**     * 发布 （模拟其他客户端发布消息）     *     * @param message 消息     */    public static void publish(String message) {        String topic = PUBLISH_TOPIC;        Integer qos = 2;        Boolean retained = false;        try {            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息            mqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());        } catch (MqttException e) {            e.printStackTrace();        }    }    /**     * 响应 （收到其他客户端的消息后，响应给对方告知消息已到达或者消息有问题等）     *     * @param message 消息     */    public void response(String message) {        String topic = RESPONSE_TOPIC;        Integer qos = 2;        Boolean retained = false;        try {            //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息            mqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());        } catch (MqttException e) {            e.printStackTrace();        }    }    /**     * 初始化     */    private void init() {        String serverURI = HOST; //服务器地址（协议+地址+端口号）        mqttAndroidClient = new MqttAndroidClient(this, serverURI, CLIENTID);        mqttAndroidClient.setCallback(mqttCallback); //设置监听订阅消息的回调        mMqttConnectOptions = new MqttConnectOptions();        mMqttConnectOptions.setCleanSession(true); //设置是否清除缓存        mMqttConnectOptions.setConnectionTimeout(10); //设置超时时间，单位：秒        mMqttConnectOptions.setKeepAliveInterval(20); //设置心跳包发送间隔，单位：秒        mMqttConnectOptions.setUserName(USERNAME); //设置用户名        mMqttConnectOptions.setPassword(PASSWORD.toCharArray()); //设置密码        // last will message        boolean doConnect = true;        String message = "{\"terminal_uid\":\"" + CLIENTID + "\"}";        String topic = PUBLISH_TOPIC;        Integer qos = 2;        Boolean retained = false;        if ((!message.equals("")) || (!topic.equals(""))) {            // 最后的遗嘱            try {                mMqttConnectOptions.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());            } catch (Exception e) {                Log.i(TAG, "Exception Occured", e);                doConnect = false;                iMqttActionListener.onFailure(null, e);            }        }        if (doConnect) {            doClientConnection();        }    }    /**     * 连接MQTT服务器     */    private void doClientConnection() {        if (!mqttAndroidClient.isConnected() && isConnectIsNomarl()) {            try {                mqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);            } catch (MqttException e) {                e.printStackTrace();            }        }    }    /**     * 判断网络是否连接     */    private boolean isConnectIsNomarl() {        ConnectivityManager connectivityManager = (ConnectivityManager) this.getApplicationContext().getSystemService(Context.CONNECTIVITY_SERVICE);        NetworkInfo info = connectivityManager.getActiveNetworkInfo();        if (info != null && info.isAvailable()) {            String name = info.getTypeName();            Log.i(TAG, "当前网络名称：" + name);            return true;        } else {            Log.i(TAG, "没有可用网络");            /*没有可用网络的时候，延迟3秒再尝试重连*/            new Handler().postDelayed(new Runnable() {                @Override                public void run() {                    doClientConnection();                }            }, 3000);            return false;        }    }    //MQTT是否连接成功的监听    private IMqttActionListener iMqttActionListener = new IMqttActionListener() {        @Override        public void onSuccess(IMqttToken arg0) {            Log.i(TAG, "连接成功 ");            try {                mqttAndroidClient.subscribe(PUBLISH_TOPIC, 2);//订阅主题，参数：主题、服务质量            } catch (MqttException e) {                e.printStackTrace();            }        }        @Override        public void onFailure(IMqttToken arg0, Throwable arg1) {            arg1.printStackTrace();            Log.i(TAG, "连接失败 ");            doClientConnection();//连接失败，重连（可关闭服务器进行模拟）        }    };    //订阅主题的回调    private MqttCallback mqttCallback = new MqttCallback() {        @Override        public void messageArrived(String topic, MqttMessage message) throws Exception {            Log.i(TAG, "收到消息： " + new String(message.getPayload()));            //收到消息，这里弹出Toast表示。如果需要更新UI，可以使用广播或者EventBus进行发送            Toast.makeText(getApplicationContext(), "messageArrived: " + new String(message.getPayload()), Toast.LENGTH_LONG).show();            //收到其他客户端的消息后，响应给对方告知消息已到达或者消息有问题等            response("message arrived");        }        @Override        public void deliveryComplete(IMqttDeliveryToken arg0) {        }        @Override        public void connectionLost(Throwable arg0) {            Log.i(TAG, "连接断开 ");            doClientConnection();//连接断开，重连        }    };    @Override    public void onDestroy() {        try {            mqttAndroidClient.disconnect(); //断开连接        } catch (MqttException e) {            e.printStackTrace();        }        super.onDestroy();    }}

该 MyMqttService 类的大概逻辑就是开启服务后，调用init()方法初始化各个参数，包括服务器地址、用户名、密码等等，然后调用doClientConnection()方法连接MQTT服务器，iMqttActionListener用来监听MQTT是否连接成功，连接成功则订阅主题。mqttCallback为订阅主题的回调，收到消息后会执行该回调中的messageArrived()方法，拿到消息后进行UI更新，并调用response()方法响应给对方告知消息已到达或者消息有问题等。

2.3 开启服务

在MainActivity中开启服务，这里为了方便不做UI更新，所以就一行开启服务的代码，如下：

import android.os.Bundle;import android.support.v7.app.AppCompatActivity;import android.view.View;public class MainActivity extends AppCompatActivity {   @Override   protected void onCreate(Bundle savedInstanceState) {       super.onCreate(savedInstanceState);       setContentView(R.layout.activity_main);       MyMqttService.startService(this); //开启服务   }}

1 集成

2 具体代码

2.1 Android中使用MQTT最主要的就是以下几个方法：

2.2 MQTT服务——MyMqttService

2.3 开启服务

更多相关文章

随机推荐