MQTT客户端实现

---

title: MQTT客户端实现

我们使用ibm提供的com.ibm.micro.client.mqttv3.jar来实现。官方教程链接如下Getting started with the MQTT client for Java on Android 鉴于英语水平有限，我还是算了吧。

来说干货。

1、首先导入com.ibm.micro.client.mqttv3.jar包 com.ibm.micro.client.mqttv3.jar

2、我们来实现MqttV3Service的类，其中里面实现了mqtt的连接、断开、发布消息的方法。

package mqtt;

import android.os.Handler;

import com.ibm.micro.client.mqttv3.MqttClient;
import com.ibm.micro.client.mqttv3.MqttConnectOptions;
import com.ibm.micro.client.mqttv3.MqttDeliveryToken;
import com.ibm.micro.client.mqttv3.MqttException;
import com.ibm.micro.client.mqttv3.MqttMessage;
import com.ibm.micro.client.mqttv3.MqttPersistenceException;
import com.ibm.micro.client.mqttv3.MqttTopic;

import java.util.ArrayList;

public class MqttV3Service {
    String addr = "";
    String port = "";
    private static MqttClient client = null;
    private static MqttTopic topic = null;
    static ArrayList<MqttTopic> topicList = new ArrayList<MqttTopic>();

    public static boolean connectionMqttServer(Handler handler, String ServAddress, String ServPort, String userID, ArrayList<String> Topics) {
        String connUrl = "tcp://" + ServAddress + ":" + ServPort;
        try {
            client = new MqttClient(connUrl, userID, null);
            for (int i = 0; i < Topics.size(); i++) {
                topic = client.getTopic(Topics.get(i));
                topicList.add((MqttTopic) topic);
            }
            CallBack callback = new CallBack(userID, handler);
            client.setCallback(callback);
            MqttConnectOptions conOptions = new MqttConnectOptions();
//          conOptions.setUserName(MyApplication.gUserName);
//          conOptions.setPassword(Pssword.toCharArray());
            conOptions.setCleanSession(false);
//          char[] ddd = conOptions.getPassword();
//          System.out.println(ddd);
            client.connect(conOptions);
            for (int i = 0; i < Topics.size(); i++) {
                client.subscribe(Topics.get(i), 1);

            }
        } catch (MqttException e) {
            // TODO 自动生成的 catch 块
            e.printStackTrace();
            return false;
        }
        return true;
    }

    public static boolean closeMqtt() {
        try {
            client.disconnect();
        } catch (MqttException e) {
            // TODO 自动生成的 catch 块
            e.printStackTrace();
            return false;
        }
        return true;
    }

    public static boolean publishMsg(String msg, int Qos, int position) {
        MqttMessage message = new MqttMessage(msg.getBytes());
        message.setQos(Qos);
        MqttDeliveryToken token;
        try {
            token = topicList.get(position).publish(message);
            while (!token.isComplete()) {
                token.waitForCompletion(1000);
            }
        } catch (MqttPersistenceException e) {
            // TODO 自动生成的 catch 块
            e.printStackTrace();
            return false;
        } catch (MqttException e) {
            // TODO 自动生成的 catch 块
            e.printStackTrace();
            return false;
        }
        return true;
    }
}

在里面我们用到了用户端的一个回调CallBack(userID, handler)我们先来看看MqttCallback接口的方法

public interface MqttCallback {
    void connectionLost(Throwable var1);

    void messageArrived(MqttTopic var1, MqttMessage var2) throws Exception;

    void deliveryComplete(MqttDeliveryToken var1);
}

由此我们可以看到，我们可以使用回调来实现消息到达的处理、连接断开、发送完毕的处理。

package mqtt;

import android.os.*;
import com.ibm.micro.client.mqttv3.*;

public class CallBack implements MqttCallback {
    private String instanceData = "";
    private Handler handler;

    public CallBack(String instance, Handler handler) {
        instanceData = instance;
        this.handler = handler;
    }

    public void messageArrived(MqttTopic topic, MqttMessage message) {
        try {
            Message msg = Message.obtain();
            Bundle bundle = new Bundle();
            bundle.putString("content", message.toString());
            msg.what = 2;
            msg.setData(bundle);
            handler.sendMessage(msg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    //下面两个方法交给你们自己去实现吧。😄
    @Override
    public void connectionLost(Throwable throwable) {

    }

    public void deliveryComplete(MqttDeliveryToken token) {

    }
}

至此我们已经完成了mqtt功能驱动。下一步我们便可以结合具体使用情境来做了

3、在Activity中结合具体方法来使用。

其实我们使用也不外乎就这么几种，连接、订阅话题、发布消息，断线处理。我们可以建立一个线程，在这个线程中配置好mqtt的连接，因为我们已经在连接的方法中封装了回调函数。那么当我们连接成功后如果有消息到达等都会接受到。我们可以使用一个handler来负责接受各种消息。思路有了，代码就有了。

    @Override
    public void onNothingSelected(AdapterView<?> parent) {

    }
    public class MqttProcThread implements Runnable
    {

        int randomid  =(int)Math.floor(10000+Math.random()*90000);
        @Override
        public void run() {
            Message msg = new Message();
            boolean ret = MqttV3Service.connectionMqttServer(myHandler,ADDRESS,PORT,"lexin"+randomid,topicList);
            if (ret){
                msg.what = 1;
            }else{
                msg.what = 0;
            }
            msg.obj = "strresult";
            myHandler.sendMessage(msg);
        }
    }
    @SuppressWarnings("HandlerLeak")
    private Handler myHandler = new Handler() {
        @Override
        public void handleMessage(Message msg) {
            super.handleMessage(msg);
            if (msg.what == 1){
                Toast.makeText(context,"连接成功",Toast.LENGTH_SHORT).show();

            }else if (msg.what == 0){
                Toast.makeText(context,"连接失败",Toast.LENGTH_SHORT).show();
            }else if(msg.what == 2) {
                String strContent = "";
                strContent += msg.getData().getString("content");
                System.out.println("strcontent:"+strContent);
            }
            else if (msg.what == 3){
                if (MqttV3Service.closeMqtt()){
                    Toast.makeText(context,"断开连接",Toast.LENGTH_SHORT).show();
                }
            }
        }
    };

如此OK。 在我们使用的地方只需新建这个线程就可以了

new Thread(new MqttProcThread()).start();

至此我们已经完成了mqtt客户端的功能了。
👏👏👏👏👏👏👏👏👏👏👏👏👏👏👏👏👏👏👏