Android MQTT管理类完整代码

Posted 2023-02-07 薛萌

mport android.annotation.SuppressLint;
import android.content.Context;
import android.text.TextUtils;

import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttManager 

    @SuppressLint("StaticFieldLeak")
    private static volatile MqttManager mInstance = null;
    private MqttCallback mCallback;
    private MqttClient client;
    private MqttConnectOptions conOpt;
    private Context context;
    private String[] topic;

    private MqttManager(Context context) 
        mCallback = new MqttCallbackBus(context);
        this.context = context;
    

    public static MqttManager getInstance(Context context) 
        if (mInstance == null) 
            synchronized (MqttManager.class) 
                if (mInstance == null) 
                    mInstance = new MqttManager(context);
                
            
        
        return mInstance;
    

    /**
     * 释放单例, 及其所引用的资源
     */
    public static void release() 
        try 
            if (mInstance != null) 
                mInstance.disConnect();
                mInstance = null;
            
         catch (Exception e) 
            e.printStackTrace();
        
    

    /**
     * 创建Mqtt 连接
     *
     * @param brokerUrl Mqtt服务器地址(tcp://xxxx:1863)
     * @param userName  用户名
     * @param password  密码
     * @return
     */
    public boolean createConnect(String brokerUrl, String userName, String password) 
        topic = new String[]Constant.getOnlineTopic();
        String deviceId = Utils.getMacStr();
        if (client != null && client.isConnected()) 
            return true;
        
        boolean flag = false;
        try 
            conOpt = new MqttConnectOptions();
            conOpt.setCleanSession(true);
            conOpt.setAutomaticReconnect(true);
            if (!TextUtils.isEmpty(password)) 
                conOpt.setPassword(password.toCharArray());
            
            if (!TextUtils.isEmpty(userName)) 
                conOpt.setUserName(userName);
            
            //Will will = new Will();
            //will.addHead(Constant.getDeviceMac(), (short) 0);
            //conOpt.setWill("/clientWill", will.encode(), 2, false);
            client = new MqttClient(brokerUrl, deviceId, new MemoryPersistence());
            client.setCallback(mCallback);
            //doConnect()和subscribe()先后顺序不能变
            flag = doConnect();
            client.subscribe(topic);
         catch (MqttException e) 
            e.printStackTrace();
        
        return flag;
    

    /**
     * 建立连接
     *
     * @return
     */
    private boolean doConnect() 
        boolean flag = false;
        if (client != null) 
            try 
                client.connect(conOpt);
                flag = true;
             catch (Exception e) 
                e.printStackTrace();
            
        
        return flag;
    

    public boolean publish(String topicName, int qos, byte[] payload) 
        boolean flag = false;
        try 
            if (client == null) 
                createConnect(Constant.MQTT_HOST, Constant.USERNAME, Constant.PASSWORD);
            
            if (!client.isConnected()) 
                this.reconnect();
            
            MqttMessage message = new MqttMessage(payload);
            message.setQos(qos);
            client.publish(topicName, message);
            flag = true;
         catch (MqttException e) 
            e.printStackTrace();
        
        return flag;
    


    private boolean subscribe(String topicName, int qos) 
        boolean flag = false;
        if (client != null && client.isConnected()) 
            try 
                client.subscribe(topicName, qos);
                flag = true;
             catch (MqttException e) 
                e.printStackTrace();
            
        
        return flag;
    

    private boolean subscribe(String[] topicName, int qos[]) 
        boolean flag = false;
        if (client != null && client.isConnected()) 
            try 
                client.subscribe(topicName, qos);
                flag = true;
             catch (MqttException e) 
                e.printStackTrace();
            
        
        return flag;
    


    /**
     * 取消连接
     *
     * @throws MqttException
     */
    public void disConnect() throws MqttException 
        if (client != null && client.isConnected()) 
            client.disconnect();
        
    

    /**
     * 关闭连接
     */
    public void close() 
        if (client != null && client.isConnected()) 
            try 
                client.disconnect();
             catch (MqttException e) 
                e.printStackTrace();
            
        
    

    /**
     * 重新连接
     */
    private void reconnect() 
        if (client != null && !client.isConnected()) 
            try 
                client.setCallback(mCallback);
                client.connect(conOpt);
                client.subscribe(topic);
             catch (MqttException e) 
                e.printStackTrace();
            
        
    

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class MqttCallbackBus implements MqttCallbackExtended 

    private Context mContext;

    public MqttCallbackBus(Context context) 
        this.mContext = context;
    

    @Override
    public void connectComplete(boolean reconnect, String serverURI) 
       
    

    @Override
    public void connectionLost(Throwable cause) 
        Log.e("MqttCallbackBus>>>", "connectionLost:");
        cause.printStackTrace();
    

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception 
        Log.e("messageArrived>>>topic:", topic);
       
    

    @Override
    public void deliveryComplete(IMqttDeliveryToken token)