Android MQTT与WebSocket协议详细讲解

 更新时间:2022年11月01日 08:31:51   作者:幸大叔  
MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在TCP/IP协议族上,是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议

MQTT

MQTT是一个极其轻量级的发布/订阅消息传输协议,对于需要较小代码占用空间或网络带宽非常宝贵的远程连接非常有用

有如下特点:

  • 开放消息协议,简单易实现;
  • 发布订阅模式,一对多消息发布;
  • 基于TCP/IP网络连接,提供有序,无损,双向连接;
  • 1字节固定报头,2字节心跳报文,最小化传输开销和协议交换,有效减少网络流量;
  • 消息QoS支持,可靠传输保证。

添加依赖

        maven { url "https://repo.eclipse.org/content/repositories/paho-snapshots/" }

    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'
    implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

首先,连接服务器,SERVER_HOST为主机名,CLIENT_ID为客户端唯一标识,还需要用户名和密码,当然,这些一般由服务器返回。

    fun connect() {
        try {
            //MemoryPersistence设置clientId的保存形式,默认为以内存保存
            client = MqttAsyncClient(SERVER_HOST, CLIENT_ID, MemoryPersistence())
            //MQTT连接设置
            options = MqttConnectOptions()
            with(options) {
                //是否清空session,true表示每次连接到服务器都以新的身份,false表示服务器会保留客户的连接记录
                isCleanSession = true
                //用户名和密码
                userName = USERNAME
                password = PASSWORD.toCharArray()
                //超时时间,单位是秒
                connectionTimeout = 30
                //会话心跳时间,单位是秒,服务器每隔30秒向客户端发送消息判断客户端是否在线
                keepAliveInterval = 30
            }
            //设置回调
            client!!.setCallback(PushCallBack())
            client!!.connect(options, context, iMqttActionListener)
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

心跳包:在空闲时间内,如果客户端没有其他任何报文发送,必须发送一个PINGREQ报文到服务器,而如果服务端在 1.5 倍心跳时间内没有收到客户端消息,则会主动断开客户端的连接,发送其遗嘱消息给所有订阅者,而服务端收到PINGREQ报文之后,立即返回PINGRESP报文给客户端。

如果两个客户端的 clientID 一样,那么服务端记录第一个客户端连接之后再收到第二个客户端连接请求,则会向第一个客户端发送 Disconnect 报文断开连接,并连接第二个客户端。

遗嘱消息:MqttConnectOptions的setWill方法可以设置遗嘱消息,客户端没有主动向服务端发起disconnect断开连接消息,但服务端检测到和客户端的连接已断开,此时服务端将该客户端设置的遗嘱消息发送出去,遗嘱Topic中不能存在通配符。

应用场景:客户端掉线之后,可以及时通知到所有订阅该遗嘱topic的客户端。

订阅消息。

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

MQTT是一种发布/订阅的消息协议, 通过设定的主题topic,发布者向topic发送的payload负载消息会经过服务器,转发到所有订阅该topic的订阅者。topic有两个通配符,“+” 和 “#”,与 “/” 组合使用,“+” 只能表示一级topic,“#” 可以表示任意层级,例如,订阅topic为 “guangdong/+”,发布者发布的topic可以是 guangdong,guangdong/shenzhen,但是不能是guangdong/shenzhen/nanshan,而订阅的topic如果是 “guangdong/#” 则可以收到。

Qos为服务质量等级,qos=0,表示发送方只发一次,不管是否发成功,也不管接收方是否成功接收,适用于不重要的数据传输;qos=1,表示消息至少有一次到达接收方,发送方发送消息,需要等待接收方返回应答消息,如果发送方在一定时间内未收到应答,发送方将继续发送,直到收到应答消息,删除本地消息缓存,不再发送。适用于需要收到所有消息,客户端可以处理重复消息;qos = 2:确保消息只一次到达接收方,一般我们都会选择2。

发布消息

    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()
        }
    }

payload为负载消息,字节流类型,是 MQTT 通信传输的真实数据。retain是保留信息,服务端将保留对应topic最新的一条消息记录,保留消息的作用是每次客户端连上都会收到其topic的最后一条保留消息。

整个封装类如下:

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
            }
        }
    }
    /**
     * 连接服务器
     */
    fun connect() {
        try {
            //MemoryPersistence设置clientId的保存形式,默认为以内存保存
            client = MqttAsyncClient(SERVER_HOST, CLIENT_ID, MemoryPersistence())
            //MQTT连接设置
            options = MqttConnectOptions()
            with(options) {
                //是否清空session,true表示每次连接到服务器都以新的身份,false表示服务器会保留客户的连接记录
                isCleanSession = true
                //用户名和密码
                userName = USERNAME
                password = PASSWORD.toCharArray()
                //超时时间,单位是秒
                connectionTimeout = 30
                //会话心跳时间,单位是秒,服务器每隔30秒向客户端发送消息判断客户端是否在线
                keepAliveInterval = 30
                //自动重连
                isAutomaticReconnect = true
            }
            //设置回调
            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()
        }
    }
    /**
     * 断开连接
     */
    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?) {
            //接收消息回调
        }
        override fun deliveryComplete(token: IMqttDeliveryToken?) {
            //发布消息完成后的回调
        }
    }
}

WebSocket

WebSocket是应用层的一种协议,是建立在TCP协议基础上的,主要特点就是全双工通信,允许服务器主动发送信息给客户端。

这里使用OKHTTP进行WebSocket开发。

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() {
            //连接成功后回调
            override fun onOpen(webSocket: WebSocket, response: Response) {
                super.onOpen(webSocket, response)
            }
            //服务器发送消息给客户端时回调
            override fun onMessage(webSocket: WebSocket, text: String) {
                super.onMessage(webSocket, text)
            }
            //服务器发送的byte类型消息
            override fun onMessage(webSocket: WebSocket, bytes: ByteString) {
                super.onMessage(webSocket, bytes)
            }
            //服务器连接关闭中
            override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
                super.onClosing(webSocket, code, reason)
            }
            //服务器连接已关闭
            override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
                super.onClosed(webSocket, code, reason)
            }
            //服务器连接失败
            override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
                super.onFailure(webSocket, t, response)
            }
        })
    }
    //发送消息
    private fun sendMessage(message: String) {
        webSocket?.send(message)
    }
    private fun close(code: Int, reason: String) {
        webSocket?.close(code, reason)
    }
}

总结

WebSocket是一种简单的报文协议,着重解决客户端与服务端不能双向通信的问题。

MQTT是基于TCP的发布/订阅消息传输协议,客户端可以创建和订阅任意主题,并向主题发布或接收消息,此外,有许多为物联网优化的特性,比如服务质量等级Qos,层级主题,遗嘱信息等。

MQTT和WebSocket都是应用层协议,都使用TCP协议来确保可靠传输,都支持双向通信,都使用二进制编码。WebSocket更简单灵活,MQTT相对复杂,但功能强大,大家可以根据自己的使用场景来选择 。

到此这篇关于Android MQTT与WebSocket协议详细讲解的文章就介绍到这了,更多相关Android MQTT与WebSocket内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

相关文章

  • Android页面中可编辑与不可编辑切换的实现

    Android页面中可编辑与不可编辑切换的实现

    这篇文章主要给大家介绍了关于在Android页面中可编辑与不可编辑切换的实现方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2018-07-07
  • Android  XML設置屏幕方向(android:screenOrientation)详解

    Android XML設置屏幕方向(android:screenOrientation)详解

    这篇文章主要介绍了Android XML設置屏幕方向(android:screenOrientation)详解的相关资料,需要的朋友可以参考下
    2017-04-04
  • Android 页面多状态布局管理的开发

    Android 页面多状态布局管理的开发

    页面多状态布局是开发中常见的需求,即页面在不同状态需要显示不同的布局,这篇文章主要介绍了Android 页面多状态布局管理的开发,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-10-10
  • Android实现横向滑动卡片效果

    Android实现横向滑动卡片效果

    这篇文章主要为大家详细介绍了Android实现横向滑动卡片效果,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下
    2018-12-12
  • Flutter 系统是如何实现ExpansionPanelList的示例代码

    Flutter 系统是如何实现ExpansionPanelList的示例代码

    Flutter组件有一个很大的特色,那就是很多复杂的组件都是通过一个一个小组件拼装而成的,今天就来说说系统的ExpansionPanelList是如何实现的,需要的朋友可以参考下
    2020-05-05
  • android几种不同对话框的实现方式

    android几种不同对话框的实现方式

    这篇文章介绍了android几种不同对话框的实现,主要包括:1、显示提示消息的对话框.2、简单列表项对话框。3、单选列表项对话框。4、多选列表对话框。5、自定义列表项对话框。6、自定义View的对话框,需要的朋友可以参考下
    2015-08-08
  • android自定义对话框实例代码

    android自定义对话框实例代码

    大家好,本篇文章主要讲的是android自定义对话框实例代码,感兴趣的同学赶快来看一看吧,对你有帮助的话记得收藏一下,方便下次浏览
    2021-12-12
  • GreenDao 3.2.0 的基本使用

    GreenDao 3.2.0 的基本使用

    本文主要对GreenDao 3.2.0 的基本使用进行详细介绍。具有一定的参考价值,下面跟着小编一起来看下吧
    2017-01-01
  • Android drawable微技巧,你不知道的drawable细节

    Android drawable微技巧,你不知道的drawable细节

    今天小编就为大家分享一篇关于Android drawable微技巧,你不知道的drawable细节,小编觉得内容挺不错的,现在分享给大家,具有很好的参考价值,需要的朋友一起跟随小编来看看吧
    2018-10-10
  • Android NDK入门初识(组件结构开发流程)

    Android NDK入门初识(组件结构开发流程)

    这篇文章主要为大家介绍了Android NDK入门之初识组件结构开发流程详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-08-08

最新评论