您的位置:首页 > 游戏 > 游戏 > 建站产品_上海建设咨询有限公司_站长工具排名查询_收录优美图片找不到了

建站产品_上海建设咨询有限公司_站长工具排名查询_收录优美图片找不到了

2024/12/25 8:25:23 来源:https://blog.csdn.net/qq_40853919/article/details/143910913  浏览:    关键词:建站产品_上海建设咨询有限公司_站长工具排名查询_收录优美图片找不到了
建站产品_上海建设咨询有限公司_站长工具排名查询_收录优美图片找不到了

目录:

  1. 问题
  2. MQTT是什么以及为什么使用
  3. 如何使用:第一阶段、基础功能
  4. 如何使用:第二阶段、增加断网重连
  5. 如何使用:第三阶段、封装

在这里插入图片描述


一、问题

在开发的时候,我们一般都使用Http和后台进行通讯,比如我们是开发物联网的,设备会有很多数据需要频繁发给后台,使用Http来做这件事情,就感觉很重,比如会遇到如下这些问题:

  1. 开发成本:需要后台创建接口,前台去请求。
  2. 连接数过多:在HTTP协议中,每次请求都需要建立一个新的连接,这可能导致连接数过多,特别是在高并发场景下。对于自动售卖机来说,如果同时有大量的用户进行交互,可能会导致服务器资源紧张,影响性能。
  3. 开销较大:HTTP协议的消息头部相对复杂,包含了大量的元数据,这增加了网络传输的开销。
  4. 实时性较差:HTTP协议是基于请求-响应模式的,需要客户端主动发起请求才能获取数据。这导致在实时性要求较高的场景下,HTTP可能无法满足需求。也就是服务器不能主动发数据给客户端。

基于这样的背景,本来想使用Rabbit MQ,但是不能双向通讯,我们选择切换成MQTT。


二、MQTT是什么以及为什么使用

MQTT(Message Queuing Telemetry Transport)是一个轻量级的发布/订阅消息协议,它构建于TCP/IP协议之上,为小型设备提供了稳定的网络通讯。MQTT协议设计简单,易于实现,非常适合在物联网(IoT)和移动应用中使用。

你会发现传递的数据量是根据你的内容来决定。

能干吗:

1、实时通讯:MQTT支持异步通讯模式,客户端可以通过订阅主题来接收感兴趣的消息,而不需要主动请求。这使得MQTT非常适合实时通讯和事件驱动的应用场景。
2、低开销:MQTT协议的数据包开销非常小,消息头部仅需2字节,非常适合网络带宽受限或设备资源受限的环境。
3、高可靠性:MQTT支持三种不同的服务质量(QoS)级别,可以根据实际需求选择合适的级别来确保消息的可靠传输。同时,MQTT还具有自动重连机制,能够在网络断开时自动恢复连接。
4、减少连接数:与HTTP相比,MQTT协议只需要客户端与服务器(Broker)建立一次连接就可以进行多次消息发布和订阅,大大减少了网络连接次数和数据传输量。


三、如何使用:第一阶段、基础功能

  1. 如何连接:init方法
  2. 连接后如何订阅:subscribe方法
  3. 如何发送数据,如何接受数据:subscribe方法
/*** 测试环境的设备管理系统*/
class ManageMqtt {private var TAG = "MQTT"private var client: MqttAndroidClient? = null //mqtt客户端private lateinit var options: MqttConnectOptions  //mqtt 的链接信息设置@Volatilevar isMqConnected: Boolean = false//初始化,fun init(context: Context?) {try {log("1")if (client != null) {return}log("1")//MQTT的连接设置options = MqttConnectOptions()//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录,这里设置为true表示每次连接到服务器都以新的身份连接options.isCleanSession = true//重连尝试options.isAutomaticReconnect = true// 设置超时时间 单位为秒options.connectionTimeout = 10// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线,但这个方法并没有重连的机制options.keepAliveInterval = 90client = MqttAndroidClient(context, "tcp://xxx:xxxx", "名称")//名称//设置连接的用户名options.userName = "xxx"//设置连接的密码options.password = "xxx".toCharArray()//设置回调client?.setCallback(object : MqttCallbackExtended {override fun connectComplete(reconnect: Boolean, serverURI: String) {log("已连接mq")isMqConnected = true//连接成功,我们要进行订阅subscribe("xxxx")}override fun connectionLost(cause: Throwable) {log("已断开mq")isMqConnected = false}override fun deliveryComplete(token: IMqttDeliveryToken) {//publish后会执行到这里  发布try {log("发送成功:" + token.message.toString())} catch (e: Exception) {e.printStackTrace()}}override fun messageArrived(topicName: String, message: MqttMessage) {//subscribe后得到的消息会执行到这里面  订阅//topicName 为主题try {//todo 收到消息,要进行一些处理的。log("收到消息:$topicName     $message")} catch (e: Exception) {log("异常:$e")}}})connect()} catch (e: Exception) {e.printStackTrace()}}//进行链接private fun connect() {Thread(connect).start()
//        Schedulers.io().scheduleDirect(connect)}private val connect = Runnable {if (client != null && client!!.isConnected) {return@Runnable}try {log("连接Mq............")client?.connect(options, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken) {log("Connection success")//todo 是否连接成功?要重连的。}override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {log("Connection failure")//todo 是否连接成功?要重连的。}})} catch (e: Exception) {e.printStackTrace()}}//订阅信息fun subscribe(topic: String, qos: Int = 1) {try {client?.subscribe(topic, qos, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken?) {Log.d(TAG, "Subscribed to $topic")}override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {Log.d(TAG, "Failed to subscribe $topic")}})} catch (e: MqttException) {e.printStackTrace()}}//发送消息fun publish(topic: String, msg: String, qos: Int = 1, retained: Boolean = false) {try {val message = MqttMessage()message.payload = msg.toByteArray()message.qos = qosmessage.isRetained = retainedclient?.publish(topic, message, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken?) {Log.d(TAG, "$msg published to $topic")}override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {Log.d(TAG, "Failed to publish $msg to $topic")}})} catch (e: MqttException) {e.printStackTrace()}}//释放资源fun closeMqtt() {try {if (client != null) {client!!.disconnect()client = null}} catch (e: java.lang.Exception) {e.printStackTrace()}}//打印logprivate fun log(msg: String) {Log.d(TAG, msg)}}

四、如何使用:第二阶段、断网重连

  1. 即使短暂断网,后面自己也还是可以重连恢复。
  2. 如果第一次没有连接上,增加第一次的断网重连
/*** 测试环境的设备管理系统*/
class ManageMqtt {private var context: Context? = nullprivate var TAG = "MQTT"private var client: MqttAndroidClient? = null //mqtt客户端private lateinit var options: MqttConnectOptions  //mqtt 的链接信息设置@Volatilevar isMqConnected: Boolean = false//初始化,fun init(context: Context?) {this.context = contexttry {log("1")if (client != null) {return}log("1")//MQTT的连接设置options = MqttConnectOptions()//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录,这里设置为true表示每次连接到服务器都以新的身份连接options.isCleanSession = true//重连尝试options.isAutomaticReconnect = true// 设置超时时间 单位为秒options.connectionTimeout = 10// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线,但这个方法并没有重连的机制options.keepAliveInterval = 90client = MqttAndroidClient(context, "tcp://xxxx:xxxx", "")//名称//设置连接的用户名options.userName = "xxx"//设置连接的密码options.password = "xxx".toCharArray()//设置回调client?.setCallback(object : MqttCallbackExtended {override fun connectComplete(reconnect: Boolean, serverURI: String) {log("已连接mq")isMqConnected = true//连接成功,我们要进行订阅subscribe("xxxx")}override fun connectionLost(cause: Throwable) {log("已断开mq")isMqConnected = false}override fun deliveryComplete(token: IMqttDeliveryToken) {//publish后会执行到这里  发布try {log("发送成功:" + token.message.toString())} catch (e: Exception) {e.printStackTrace()}}override fun messageArrived(topicName: String, message: MqttMessage) {//subscribe后得到的消息会执行到这里面  订阅//topicName 为主题try {//todo 收到消息,要进行一些处理的。 Eventbuslog("收到消息:$topicName     $message")} catch (e: Exception) {log("异常:$e")}}})connect()} catch (e: Exception) {e.printStackTrace()}val intentFilter = IntentFilter()intentFilter.addAction(ConnectivityManager.CONNECTIVITY_ACTION)intentFilter.addAction(WifiManager.NETWORK_STATE_CHANGED_ACTION)intentFilter.addAction(WifiManager.WIFI_STATE_CHANGED_ACTION)intentFilter.addAction(WifiManager.RSSI_CHANGED_ACTION)context?.registerReceiver(netWorkBroadCastReciver,intentFilter)}//进行链接private fun connect() {Thread(connect).start()
//        Schedulers.io().scheduleDirect(connect)}private val connect = Runnable {if (client != null && client!!.isConnected) {return@Runnable}try {log("连接Mq............")client?.connect(options, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken) {log("Connection success")//todo 是否连接成功?要重连的。}override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {log("Connection failure")//todo 是否连接成功?要重连的。}})} catch (e: Exception) {e.printStackTrace()}}//订阅信息fun subscribe(topic: String, qos: Int = 1) {try {client?.subscribe(topic, qos, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken?) {Log.d(TAG, "Subscribed to $topic")}override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {Log.d(TAG, "Failed to subscribe $topic")}})} catch (e: MqttException) {e.printStackTrace()}}//发送消息/*** @param topic 主题 给这个主题发送消息*  @param qos 0最多一次不管是否收到,1最少一次可能会收到多次,2保证收到,且仅一次*  @param retained 发布后是否保留,即重新链接时会存在*  @param msg 消息*/fun publish(topic: String, msg: String, qos: Int = 0, retained: Boolean = false) {try {val message = MqttMessage()message.payload = msg.toByteArray()message.qos = qosmessage.isRetained = retained //发布后是否保留,即重新链接时会存在client?.publish(topic, message, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken?) {Log.d(TAG, "$msg published to $topic")}override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {Log.d(TAG, "Failed to publish $msg to $topic")}})} catch (e: MqttException) {e.printStackTrace()}}//释放资源fun closeMqtt() {try {if (client != null) {client!!.disconnect()client = null}} catch (e: java.lang.Exception) {e.printStackTrace()}context?.unregisterReceiver(netWorkBroadCastReciver)}//打印logprivate fun log(msg: String) {Log.d(TAG, msg)}private var networkState = 100//断网重连查询fun isNetConnected(context: Context): Boolean {Log.d(TAG, "isNetConnected: ")val connectivity =context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManagerif (connectivity != null) {val info = connectivity.activeNetworkInfoif (info != null) {if (info.type == networkState) {return false}networkState = info.typeif (info.type == (ConnectivityManager.TYPE_WIFI)) {if (!isMqConnected) {connect()}return true} else if (info.type == (ConnectivityManager.TYPE_MOBILE)) {if (!isMqConnected) {connect()}return true}}}return false}var netWorkBroadCastReciver = object :BroadcastReceiver() {override fun onReceive(context: Context, intent: Intent?) {isNetConnected(context)log( "NetWorkBroadCastReciver: ")}}}

五、如何使用:第三阶段、封装

  1. 尝试封装,其实就是提供比如,注册,取消注册,订阅,发送数据,或者读取数据的方法。后面如何更换MQTT为其他协议,也很方便
/*** 对Mqtt操作的进一步封装*/
@Singleton
class MqttHelper @Inject constructor() {@Injectlateinit var mqtt: ManageMqtt/*** 注册*/fun register(context: Context?){mqtt.init(context)}/*** 发送数据*/fun sendData(data :String){Heartbeat.deviceId?.let { mqtt.publish(it,Gson().toJson(data)) }}/*** 接收数据*/fun data(kind:String,data:String){//待定,一般都是通过eventbus来解决。}
}

好了,这篇文章就介绍到这里~,我是前期后期,如果你也有相关的问题,也可以在评论区讨论哦,我们下一篇文章再见。

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com