Android MQTT的发布与订阅

一、MQTT介绍

链接1（菜鸟教程）：https://www.runoob.com/w3cnote/mqtt-intro.html

连接2（MQTT中文网）：http://mqtt.p2hp.com/

连接3（Android开发之Mqtt的使用）：https://blog.csdn.net/asjqkkkk/article/details/80714234

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议）。一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议。构建于TCP/IP协议上，由IBM在1999年发布。

二、程序示例

public class MqttManager {

    private static boolean initFirst = true;//是否第一次初始化mqtt标识符
    private static String host = "tcp://47.106.172.221:8081";
    private static String userName; //mqtt用户名
    private static String passWord; //mqtt登陆密码
    private static MqttManager manager;
    private static MqttClient mqttClient;
    private static MqttConnectOptions options;
    private static String topic;//订阅的主题
    private static String clientId; //客户端id

    private static Handler handler = new Handler() {
        @Override
        public void handleMessage(Message msg) {
            super.handleMessage(msg);
            if (msg.what == 1) {
                KLog.d(msg.obj);
                EventBus.getDefault()
                        .post(new MessageEventBean(AppConstants.MQTT_EVENT_TYPE, (String) msg.obj));
            } else if (msg.what == 2) {
                KLog.d("连接成功");
                try {
                    KLog.d("订阅的主题:" + topic);
                    mqttClient.subscribe(topic, 0);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            } else if (msg.what == 3) {
                KLog.d("连接失败，系统正在重连");
            }
        }
    };

    private MqttManager() {

    }
    private static MqttCallback myMqttCallback = new MqttCallback(){

        @Override
        public void messageArrived(String topic, MqttMessage message){
            //subscribe后得到的消息会执行到这里面
            KLog.d("messageArrived topic:"+topic);
            Message msg = new Message();
            msg.what = 1;
            msg.obj = message.toString();
            handler.sendMessage(msg);
        }
        @Override
        public void connectionLost(Throwable cause) {
            KLog.d("connectionLost cause = "+cause);
            //连接丢失后，一般在这里面进行重连
                try{
                    KLog.d("mqtt重连");
                    manager.startReconnect();
                }catch (Exception e){
                    KLog.d("Exception = "+ e);
                    e.printStackTrace();
                }
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            //publish后会执行到这里
            KLog.d("deliveryComplete");
        }
    };
    private ScheduledExecutorService scheduler;

    public static MqttManager getInstance() {
        if (manager == null) {
            manager = new MqttManager();
        }
        return manager;
    }

    public void initConnection() {
        if (initFirst){
            KLog.d("第一次调用initConnection");
            try {
                clientId = Preferences.getUserAccount() + System.currentTimeMillis();//客户端标识符(本机mac地址+当前时间ms)
                userName = Preferences.getUserAccount();//用户名
                passWord = Preferences.getUserToken();//密码
                topic = userName;
                //host为主机名；clientid即连接MQTT的客户端ID，是客户端的唯一标识符；MemoryPersistence设置clientid的保存形式，默认为以内存保存
                mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
                //MQTT的连接设置
                options = new MqttConnectOptions();
                //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
                options.setCleanSession(true);
                //断开后，是否自动连接
                options.setAutomaticReconnect(true);
                //设置连接的用户名
                options.setUserName(userName);
                //设置连接的密码
                options.setPassword(passWord.toCharArray());
                // 设置超时时间 单位为秒
                options.setConnectionTimeout(10);
                // 设置会话心跳时间 单位为秒 服务器会每隔1.5*(20)秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
                options.setKeepAliveInterval(20);
                //setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息
//            options.setWill(topic,"close".getBytes(),2,true);
                //设置回调
                mqttClient.setCallback(myMqttCallback);
                KLog.d("clientId: "+clientId +", userName: "+userName+", passWord: "+passWord+", topic: "+topic);
                //设置标识符状态
                initFirst = false;
                //mqtt第一次连接
                manager.startReconnect();
            } catch (Exception e) {
                KLog.d("initConnection Exception: " + e);
                e.printStackTrace();
            }
        }else {
            KLog.d("网络重连后调用initConnection");
            manager.startReconnect();
        }
    }

    public  void startReconnect() {

        if (NetworkUtils.isConnected()){

            if (!mqttClient.isConnected()) {
                //重新连接
                connect();
                KLog.d("mqtt连接结束");
            }else {
                KLog.d("mqttClient.isConnected");
            }
//            scheduler = Executors.newSingleThreadScheduledExecutor();
//            scheduler.scheduleAtFixedRate(new Runnable() {
//                @Override
//                public void run() {
//
//                    if (!mqttClient.isConnected()) {
//                        connect();
//                        KLog.d("mqtt连接结束");
//                    }
//                }
//            }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
        }else {
            KLog.d("网络不可用");
//            scheduler.shutdown();
        }

    }

    public void sendMsg(String msg) {
        KLog.d("sendMsg");
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                KLog.d("发送的主题:" + Preferences.getUserAccount());
                String topic = Preferences.getUserAccount();
                KLog.d(topic);
                byte[] msgBytes = msg.getBytes();
                KLog.d("0000");
                mqttClient.publish(topic, msgBytes, 0, false);
                KLog.d("11111111111111111");
            } catch (MqttException e) {
                KLog.d(e);
            }
        }
    }

    //发布的主题设为pubTopic = "owh" + Preferences.getUserAccount();
    //发布主题（发布主题和订阅主题应设为不同值）
    public void publish(String topicName, String payload) {
        if (mqttClient != null && mqttClient.isConnected()) {
            // 创建和配置一个消息
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setPayload(payload.getBytes());
            message.setQos(0);
            try {
                KLog.d("1111");
                mqttClient.publish(topicName, message);
                KLog.d("2222");
            } catch (MqttException e) {
                KLog.d("publish : " + e.toString());
            }
        }
    }

    private void connect() {

        ThreadPoolManager.getInstance().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    mqttClient.connect(options);
                    Message msg = Message.obtain();
                    msg.what = 2;
                    handler.sendMessage(msg);
                } catch (Exception e) {
                    e.printStackTrace();
                    Message msg = Message.obtain();
                    msg.what = 3;
                    handler.sendMessage(msg);
                }
            }
        });

//        new Thread(new Runnable() {
//
//            @Override
//            public void run() {
//                try {
//                    mqttClient.connect(options);
//                    Message msg = new Message();
//                    msg.what = 2;
//                    handler.sendMessage(msg);
//                } catch (Exception e) {
//                    e.printStackTrace();
//                    Message msg = new Message();
//                    msg.what = 3;
//                    handler.sendMessage(msg);
//                }
//            }
//        }).start();
    }

    //断开连接
    public static void mqttDisconnect(){
        if(mqttClient !=null && mqttClient.isConnected()){
            try{
                mqttClient.disconnect();
            }catch (MqttException e){
                KLog.d("mqtt disconnect error");
                e.printStackTrace();
            }
        }
    }

}

三、注意事项

1、MQTT的客户端id（clientId）须唯一。在此项目中clientId = 本机mac地址 + 当前时间(ms)。

2、一个客户端的一个MQTT连接最好只new一个对象，避免一台设备产生多个客户端账号。

当多个发布（/订阅）的clientId相同时，会发生Mqtt反复重连的现象，无法正常发送或接收消息。

当多个发布（/订阅）的clientId不同时，会造成一台设备多个Mqtt账号同时在线，占用了多余的服务器资源。

3、一个客户端的发布Topic和订阅Topic不应相同。