fun connect() {
        try {
            //MemoryPersistence設(shè)置clientId的保存形式，默認(rèn)為以內(nèi)存保存
            client = MqttAsyncClient(SERVER_HOST, CLIENT_ID, MemoryPersistence())
            //MQTT連接設(shè)置
            options = MqttConnectOptions()
            with(options) {
                //是否清空session，true表示每次連接到服務(wù)器都以新的身份，false表示服務(wù)器會(huì)保留客戶的連接記錄
                isCleanSession = true
                //用戶名和密碼
                userName = USERNAME
                password = PASSWORD.toCharArray()
                //超時(shí)時(shí)間，單位是秒
                connectionTimeout = 30
                //會(huì)話心跳時(shí)間，單位是秒，服務(wù)器每隔30秒向客戶端發(fā)送消息判斷客戶端是否在線
                keepAliveInterval = 30
            }
            //設(shè)置回調(diào)
            client!!.setCallback(PushCallBack())
            client!!.connect(options, context, iMqttActionListener)
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    fun subscribeMessage(topic: String, qos: Int) {
        client?.let {
            try {
                it.subscribe(topic, qos)
            } catch (e: MqttException) {
                e.printStackTrace()
            }
        }
    }

    fun publish(topic: String, msg: String, isRetained: Boolean, qos: Int) {
        try {
            client?.let {
                val message = MqttMessage()
                message.qos = qos
                message.isRetained = isRetained
                message.payload = msg.toByteArray()
                it.publish(topic, message)
            }
        } catch (e: MqttPersistenceException) {
            e.printStackTrace()
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

class MQTTManager private constructor(private val context: Context) {
    private var client: MqttAsyncClient? = null
    private lateinit var options: MqttConnectOptions
    private val iMqttActionListener = object : IMqttActionListener {
        override fun onSuccess(asyncActionToken: IMqttToken?) {
            //連接成功，開始訂閱
            subscribeMessage(TOPIC, 2)
        }
        override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
            //連接失敗
        }
    }
    companion object {
        @Volatile
        private var instance: MQTTManager? = null
        fun getInstance(context: Context): MQTTManager = instance ?: synchronized(this) {
            instance ?: MQTTManager(context).also {
                instance = it
            }
        }
    }
    /**
     * 連接服務(wù)器
     */
    fun connect() {
        try {
            //MemoryPersistence設(shè)置clientId的保存形式，默認(rèn)為以內(nèi)存保存
            client = MqttAsyncClient(SERVER_HOST, CLIENT_ID, MemoryPersistence())
            //MQTT連接設(shè)置
            options = MqttConnectOptions()
            with(options) {
                //是否清空session，true表示每次連接到服務(wù)器都以新的身份，false表示服務(wù)器會(huì)保留客戶的連接記錄
                isCleanSession = true
                //用戶名和密碼
                userName = USERNAME
                password = PASSWORD.toCharArray()
                //超時(shí)時(shí)間，單位是秒
                connectionTimeout = 30
                //會(huì)話心跳時(shí)間，單位是秒，服務(wù)器每隔30秒向客戶端發(fā)送消息判斷客戶端是否在線
                keepAliveInterval = 30
                //自動(dòng)重連
                isAutomaticReconnect = true
            }
            //設(shè)置回調(diào)
            client!!.setCallback(PushCallBack())
            client!!.connect(options, context, iMqttActionListener)
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }
    /**
     * 訂閱消息
     */
    fun subscribeMessage(topic: String, qos: Int) {
        client?.let {
            try {
                it.subscribe(topic, qos)
            } catch (e: MqttException) {
                e.printStackTrace()
            }
        }
    }
    /**
     * 發(fā)布消息
     */
    fun publish(topic: String, msg: String, isRetained: Boolean, qos: Int) {
        try {
            client?.let {
                val message = MqttMessage()
                message.qos = qos
                message.isRetained = isRetained
                message.payload = msg.toByteArray()
                it.publish(topic, message)
            }
        } catch (e: MqttPersistenceException) {
            e.printStackTrace()
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }
    /**
     * 斷開連接
     */
    fun disconnect() {
        client?.takeIf {
            it.isConnected
        }?.let {
            try {
                it.disconnect()
                instance = null
            } catch (e: MqttException) {
                e.printStackTrace()
            }
        }
    }
    /**
     * 判斷是否連接
     */
    fun isConnected() = if (client != null) client!!.isConnected else false
    inner class PushCallBack : MqttCallback {
        override fun connectionLost(cause: Throwable?) {
            //斷開連接
        }
        override fun messageArrived(topic: String?, message: MqttMessage?) {
            //接收消息回調(diào)
        }
        override fun deliveryComplete(token: IMqttDeliveryToken?) {
            //發(fā)布消息完成后的回調(diào)
        }
    }
}

class WebSocketManager private constructor() {
    private val client: OkHttpClient = OkHttpClient.Builder().writeTimeout(5, TimeUnit.SECONDS)
        .readTimeout(5, TimeUnit.SECONDS)
        .connectTimeout(5, TimeUnit.SECONDS)
        .build()
    private var webSocket: WebSocket? = null
    companion object {
        val instance: WebSocketManager by lazy(mode = LazyThreadSafetyMode.SYNCHRONIZED) { WebSocketManager() }
    }
    fun connect(url: String) {
        webSocket?.cancel()
        val request = Request.Builder().url(url).build()
        webSocket = client.newWebSocket(request, object : WebSocketListener() {
            //連接成功后回調(diào)
            override fun onOpen(webSocket: WebSocket, response: Response) {
                super.onOpen(webSocket, response)
            }
            //服務(wù)器發(fā)送消息給客戶端時(shí)回調(diào)
            override fun onMessage(webSocket: WebSocket, text: String) {
                super.onMessage(webSocket, text)
            }
            //服務(wù)器發(fā)送的byte類型消息
            override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
                super.onMessage(webSocket, bytes)
            }
            //服務(wù)器連接關(guān)閉中
            override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
                super.onClosing(webSocket, code, reason)
            }
            //服務(wù)器連接已關(guān)閉
            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                super.onClosed(webSocket, code, reason)
            }
            //服務(wù)器連接失敗
            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                super.onFailure(webSocket, t, response)
            }
        })
    }
    //發(fā)送消息
    private fun sendMessage(message: String) {
        webSocket?.send(message)
    }
    private fun close(code: Int, reason: String) {
        webSocket?.close(code, reason)
    }
}