【Android】MQTT入门——服务器部署与客户端搭建

目录

• MQTT 协议
• • 简介
  • 应用场景
  • 优点
  • 缺点
  • 部署服务端
  • • 下载安装包
    • 启动服务器
    • 搭建客户端
    • • 下载SDK
      • 添加依赖
      • 配置MQTT服务和权限
      • 建立连接
      • 订阅主题
      • 发布消息
      • 取消订阅
      • 断开连接
      • MQTT客户端工具
      • 最终效果
      • 实现传感器数据采集与监测功能思路

        MQTT 协议

        简介

        MQTT（Message Queuing Telemetry Transport）是一种基于发布/订阅模式的轻量级消息传输协议，专门针对低带宽、和不稳定网络环境的物联网应用而设计，它可以用极少的代码为互联网设备提供实时可靠的消息服务。

        应用场景

        MQTT 协议主要用于物联网和移动设备等资源有限的场景中，其中包括物联网、移动互联网、智能硬件、车联网、智慧城市、远程医疗、电力、石油与能源等领域。案例例如：智能家居设备的控制与监控、传感器数据采集与监测、安防监控系统等。

        优点

        • 是一种低带宽占用的即时通讯协议，在低带宽和不稳定的网络环境中也能较好地工作。
        • 针对低带宽网络，低计算能力的设备，做了特殊的优化，使得其能适应各种物联网应用场景。
        • 简单轻量，相比于其他传输协议，其代码占用内存少，对硬件要求低，适用于受限环境和低带宽网络。
        • 通过发布-订阅架构，实现了实时可靠的消息服务，并且具有 QoS 服务质量，能够保证消息的可靠传输。

          QoS：是MQTT中消息传输的可靠性等级，由低到高分别是：

          • QoS0：最多一次。消息尽力发送一次，但不保证一定会被接收。
          • QoS1：至少一次。消息至少发送一次，但可能会被接收多次。
          • QoS2：有且仅有一次。消息只发送一次，且只会被接收一次

            缺点

            • 不提供扩展性的支持，需要使用其他技术来实现分布式消息传递和扩展性。
            • 没有提供加密和身份验证机制。如果需要安全通信，则需要使用TLS/SSL等其他协议。
            • MQTT是一种轻量级的协议，但在高并发和大规模消息传递的环境中，可能会面临性能瓶颈。此外，在使用较高的QoS级别时，可能会导致更多的网络流量和延迟，从而影响系统的性能。

              分布式：是一种计算和数据处理的方式，将计算任务或数据分散到多个计算机或节点中进行处理，可以提供高性能、高可用性和弹性的计算和数据处理能力，满足不同规模和复杂度的应用需求。

              TLS/SSL：详情请看文章👉 SSL/TLS 整理

              部署服务端

              话说在前，本文使用的是 EMQ 集成好的 MQTT 协议，使用起来相当简单，由于我的服务器即将到期，又没有续费的打算，我这里是在我的电脑上部署，如果你只是想着体验体验 MQTT ，跟着搞，准没错。

              下载安装包

              MQTT 支持部署在 Docker、Debian、Ubuntu、CentOS/RHEL、macOS、Kubernetes、Windows以及源码编译安装，小编电脑使用的是 Windows 系统，使用 Windows 下载 方式👉 前往下载所有版本的EMQX

              如果没有服务器，你可以考虑部署在你的自用电脑上。

              启动服务器

              Windows 版本的 EMQX 下载完成后，解压文件，打开解压后的 bin 目录，使用 CMD 命令启动 EMQX。

              # 解压后的 emqx 文件位于 D:\Softwares\emqx-5.2.0-windows-amd64\bin 目录下，因此使用 CMD 命令如下
              D:\Softwares\emqx-5.2.0-windows-amd64\bin>emqx start
              

              PS：EMQX 是部署在自用电脑上，当电脑重启时，需要重新启动 EMQX 服务。

              完成这一步，打开浏览器，在地址栏输入 http:// 服务器IP :18083/ ，能打开服务器网页即完成部署。

              

              搭建客户端

              下载SDK

              客户端，小编使用 Android 的Kotlin进行开发，MQTT 客户端 SDK 截止目前，已支持除 Android 外的 16 种 SDK，可根据自身需求不同，选择适合自身需求的开发包 👉 MQTT SDKs 下载

              

              添加依赖

              打开项目的 build.gradle 文件，添加 Eclipse Paho Java Client 和 Eclipse Paho Android Service 依赖到 dependencies ，并 Sync Now。

              dependencies {
                  implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.4'
                  implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1' 
              }
              

              配置MQTT服务和权限

              打开项目的 AndroidManifest.xml 文件，加入以下配置。

              	
              	
              	
              	
              	
              	mqttClient = MqttAndroidClient(context, serverURI, "kotlin_client")
              	mqttClient.setCallback(object : MqttCallback {
              	    override fun messageArrived(topic: String?, message: MqttMessage?) {
              	        Log.d(TAG, "Receive message: ${message.toString()} from topic: $topic")
              	    }
              	
              	    override fun connectionLost(cause: Throwable?) {
              	        Log.d(TAG, "Connection lost ${cause.toString()}")
              	    }
              	
              	    override fun deliveryComplete(token: IMqttDeliveryToken?) {
              	
              	    }
              	})
              	val options = MqttConnectOptions()
              	options.userName = username
              	options.password = password.toCharArray()
              	
                  mqttClient.connect(options, null, object : IMqttActionListener {
              	    override fun onSuccess(asyncActionToken: IMqttToken?) {
              	        Log.d(TAG, "Connection success")
              	    }
              	
              	    override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
              	        Log.d(TAG, "Connection failure")
              	    }
              	})
              }
              
                  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")
                  }
              })
              
                  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")
                  }
              })
              
                  override fun onSuccess(asyncActionToken: IMqttToken?) {
                      Log.d(TAG, "Unsubscribed to $topic")
                  }
                  override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                      Log.d(TAG, "Failed to unsubscribe $topic")
                  }
              })
              
                  override fun onSuccess(asyncActionToken: IMqttToken?) {
                      Log.d(TAG, "Disconnected")
                  }
                  override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                      Log.d(TAG, "Failed to disconnect")
                  }
              })