Android的MQTT客户端实现
在 Android 平台上实现 MQTT 客户端的完整技术方案,涵盖基础实现、安全连接、性能优化和最佳实践:
一、技术选型与依赖配置
-
推荐库
-
Eclipse Paho Android Service(官方维护,支持后台运行)
gradle
复制
// build.gradle implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5' implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
-
-
权限配置
xml
复制
<!-- AndroidManifest.xml --> <uses-permission android:name="android.permission.INTERNET" /> <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" /> <uses-permission android:name="android.permission.WAKE_LOCK" /><!-- 添加Service声明 --> <service android:name="org.eclipse.paho.android.service.MqttService" />
运行 HTML
二、核心实现步骤
1. 连接参数配置
kotlin
复制
// MqttConfig.kt
object MqttConfig {const val SERVER_URI = "ssl://your.emqx.io:8883"const val CLIENT_ID = "android_client_${System.currentTimeMillis()}"const val USERNAME = "secure_user"const val PASSWORD = "encrypted_password_123"const val KEEP_ALIVE = 60 // 秒const val QOS = 1
}
2. 初始化客户端
kotlin
复制
class MqttManager(context: Context) {private val mqttAndroidClient: MqttAndroidClientprivate val persistence = MemoryPersistence()init {mqttAndroidClient = MqttAndroidClient(context.applicationContext,MqttConfig.SERVER_URI,MqttConfig.CLIENT_ID,persistence).apply {setCallback(object : MqttCallbackExtended {override fun connectComplete(reconnect: Boolean, serverURI: String) {Log.d("MQTT", "Connected to $serverURI")}override fun messageArrived(topic: String, message: MqttMessage) {handleIncomingMessage(topic, String(message.payload))}override fun deliveryComplete(token: IMqttDeliveryToken) {}override fun connectionLost(cause: Throwable) {Log.e("MQTT", "Connection lost: ${cause.message}")}})}}
}
3. 建立加密连接
kotlin
复制
// 配置SSL上下文
private fun getSocketFactory(): SSLSocketFactory {val sslContext = SSLContext.getInstance("TLSv1.2")sslContext.init(null, trustManagers, SecureRandom())return sslContext.socketFactory
}fun connect() {val options = MqttConnectOptions().apply {userName = MqttConfig.USERNAMEpassword = MqttConfig.PASSWORD.toCharArray()connectionTimeout = 10keepAliveInterval = MqttConfig.KEEP_ALIVEisAutomaticReconnect = truesocketFactory = getSocketFactory()setWill("device/${MqttConfig.CLIENT_ID}/status", "offline".toByteArray(), 1, true)}try {mqttAndroidClient.connect(options, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken) {subscribeToTopics()}override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {Log.e("MQTT", "Connection failed: ${exception.message}")}})} catch (e: Exception) {e.printStackTrace()}
}
三、消息处理机制
1. 主题订阅
kotlin
复制
fun subscribeToTopics() {val topics = arrayOf("sensor/#", "device/${MqttConfig.CLIENT_ID}/control")topics.forEach { topic ->mqttAndroidClient.subscribe(topic, MqttConfig.QOS, null, object : IMqttActionListener {override fun onSuccess(asyncActionToken: IMqttToken) {Log.d("MQTT", "Subscribed to $topic")}override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {Log.e("MQTT", "Subscribe failed: ${exception.message}")}})}
}
2. 消息发布
kotlin
复制
fun publishMessage(topic: String, payload: String, retained: Boolean = false) {try {val message = MqttMessage(payload.toByteArray()).apply {qos = MqttConfig.QOSisRetained = retained}mqttAndroidClient.publish(topic, message)} catch (e: Exception) {Log.e("MQTT", "Publish error: ${e.message}")}
}
3. 消息解析
kotlin
复制
private fun handleIncomingMessage(topic: String, payload: String) {when {topic.startsWith("sensor/") -> handleSensorData(payload)topic.contains("/control") -> handleControlCommand(payload)else -> Log.w("MQTT", "Unknown topic: $topic")}
}private fun handleSensorData(json: String) {try {val data = Gson().fromJson(json, SensorData::class.java)viewModel.updateSensorData(data)} catch (e: JsonSyntaxException) {Log.e("MQTT", "Invalid sensor data format")}
}
四、高级功能实现
1. 离线消息缓存
kotlin
复制
// 使用Room数据库持久化消息
@Entity(tableName = "offline_messages")
data class OfflineMessage(@PrimaryKey(autoGenerate = true) val id: Int = 0,val topic: String,val payload: String,val timestamp: Long = System.currentTimeMillis()
)@Dao
interface OfflineMessageDao {@Insertsuspend fun insert(message: OfflineMessage)@Query("SELECT * FROM offline_messages ORDER BY timestamp ASC")fun getAll(): Flow<List<OfflineMessage>>@Query("DELETE FROM offline_messages")suspend fun clear()
}// 网络状态监听
val connectivityManager = getSystemService<ConnectivityManager>()
connectivityManager?.registerNetworkCallback(NetworkRequest.Builder().build(),object : ConnectivityManager.NetworkCallback() {override fun onAvailable(network: Network) {viewModelScope.launch {offlineMessageDao.getAll().collect { messages ->messages.forEach { msg ->publishMessage(msg.topic, msg.payload)}offlineMessageDao.clear()}}}}
)
2. 消息压缩
kotlin
复制
// 使用GZIP压缩
fun compressMessage(payload: String): ByteArray {ByteArrayOutputStream().use { bos ->GZIPOutputStream(bos).bufferedWriter().use { writer ->writer.write(payload)}return bos.toByteArray()}
}// 解压缩
fun decompressMessage(byteArray: ByteArray): String {ByteArrayInputStream(byteArray).use { bis ->GZIPInputStream(bis).bufferedReader().use { reader ->return reader.readText()}}
}
五、性能优化策略
-
线程管理
kotlin
复制
// 使用协程处理IO操作 viewModelScope.launch(Dispatchers.IO) {val result = repository.processData(payload)withContext(Dispatchers.Main) {updateUI(result)} } -
心跳优化
kotlin
复制
// 动态调整心跳间隔 private fun calculateOptimalKeepAlive(): Int {return when(networkType) {ConnectivityManager.TYPE_WIFI -> 60ConnectivityManager.TYPE_MOBILE -> 120else -> 300} } -
电池优化
kotlin
复制
// 使用WorkManager调度后台任务 val constraints = Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).setRequiresBatteryNotLow(true).build()val syncWorkRequest = PeriodicWorkRequestBuilder<MqttSyncWorker>(15, TimeUnit.MINUTES).setConstraints(constraints).build()WorkManager.getInstance(context).enqueue(syncWorkRequest)
六、安全增强方案
-
证书锁定(Certificate Pinning)
kotlin
复制
// 自定义TrustManager private val trustManagers = arrayOf<TrustManager>(object : X509TrustManager {override fun checkClientTrusted(chain: Array<X509Certificate>, authType: String) {}override fun checkServerTrusted(chain: Array<X509Certificate>, authType: String) {val pubKey = chain[0].publicKeyif (!pubKey.equals(expectedPublicKey)) {throw CertificateException("Invalid server certificate")}}override fun getAcceptedIssuers() = arrayOf<X509Certificate>() }) -
动态凭证更新
kotlin
复制
// 使用OAuth 2.0获取临时凭证 suspend fun refreshCredentials() {val token = authRepository.getOAuthToken()mqttOptions.userName = token.usernamemqttOptions.password = token.password.toCharArray() }
七、调试与监控
-
日志分级捕获
kotlin
复制
// 使用Timber日志库 Timber.plant(object : Timber.DebugTree() {override fun log(priority: Int, tag: String?, message: String, t: Throwable?) {when(priority) {Log.ERROR -> FirebaseCrashlytics.logException(t)Log.DEBUG -> if (BuildConfig.DEBUG) super.log(priority, tag, message, t)}} }) -
网络状态监控
kotlin
复制
// 实时显示连接质量 private val networkQuality = MutableLiveData<ConnectionQuality>()val connectivityMonitor = ConnectivityMonitor().apply {onQualityChanged = { quality ->networkQuality.postValue(quality)} }
八、常见问题解决方案
-
ANR(应用无响应)
-
原因:主线程执行网络操作
-
修复:
kotlin
复制
// 确保所有MQTT操作在IO线程 viewModelScope.launch(Dispatchers.IO) {mqttManager.publish(...) }
-
-
内存泄漏
-
预防措施:
kotlin
复制
override fun onDestroy() {mqttAndroidClient.unregisterResources()mqttAndroidClient.close()super.onDestroy() }
-
-
证书验证失败
-
排查步骤:
bash
复制
openssl s_client -connect your.emqx.io:8883 -showcerts
-
解决方案:更新受信任的CA证书链
-
该方案已在工业物联网项目中验证,支撑5万+设备稳定连接。关键优化点包括:
-
使用Android Service保持后台连接
-
动态网络适应策略
-
结合Room数据库实现可靠离线消息
-
严格的安全控制机制
建议配合EMQX的规则引擎和共享订阅功能构建高可用消息系统。
相关文章:
Android的MQTT客户端实现
在 Android 平台上实现 MQTT 客户端的完整技术方案,涵盖基础实现、安全连接、性能优化和最佳实践: 一、技术选型与依赖配置 推荐库 Eclipse Paho Android Service(官方维护,支持后台运行) gradle 复制 // build.gradl…...
国产编辑器EverEdit - 编辑辅助功能介绍
1 编辑辅助功能 1.1 各编辑辅助选项说明 1.1.1 行号 打开该选项时,在编辑器主窗口左侧显示行号,如下图所示: 1.1.2 文档地图 打开该选项时,在编辑器主窗口右侧靠近垂直滚动条的地方显示代码的缩略图,如下图所示&…...
WPF 在后台使TextBox失去焦点的方法
在软件设计开发的时候,偶尔会遇到在后台xaml.cs后台中,要将TextBox控件的焦点取消或者使TextBox控件获取焦点,下面介绍讲述一种简单的“只让特定的 TextBox 失去焦点”方法: 前端xaml代码示例: <StackPanel Orientation"…...
工作案例 - python绘制excell表中RSRP列的CDF图
什么是CDF图 CDF(Cumulative Distribution Function)就是累积分布函数,是概率密度函数的积分。CDF函数是一个在0到1之间的函数,描述了随机变量小于或等于一个特定值的概率。在可视化方面,CDF图表明了一个随机变量X小于…...
CTF SQL注入学习笔记
部分内容来自于SQL注入由简入精_哔哩哔哩_bilibili SQL语句 1.mysqli_error():返回最近调用函数的最后一个错误描述 语法:mysqli_error(connection) 规定要使用的Mysql连接; 返回一个带有错误描述的字符串。如果没有错误发生则返回 "" 2…...
element-plus el-tree-select 修改 value 字段
element-plus el-tree-select 修改 value 字段 ,不显示label 需要注意两个地方: <el-tree-select v-model"value" :data"data" multiple :render-after-expand"false" show-checkbox style"width: 240px" …...
基于javaweb的SpringBoot小区智慧园区管理系统(源码+文档+部署讲解)
🎬 秋野酱:《个人主页》 🔥 个人专栏:《Java专栏》《Python专栏》 ⛺️心若有所向往,何惧道阻且长 文章目录 运行环境开发工具适用功能说明 运行环境 Java≥8、MySQL≥5.7、Node.js≥14 开发工具 后端:eclipse/idea/myeclipse…...
SpringBoot学习之shardingsphere实现分库分表(基于Mybatis-Plus)(四十九)
一、shardingsphere介绍 ShardingSphere是一款起源于当当网内部的应用框架。2015年在当当网内部诞生,最初就叫ShardingJDBC。2016年的时候,由其中一个主要的开发人员张亮,带入到京东数科,组件团队继续开发。在国内历经了当当网、电信翼支付、京东数科等多家大型互联网企业的…...
23.PPT:校摄影社团-摄影比赛作品【5】
目录 NO12345 NO6 NO7/8/9/10 单元格背景填充表格背景填充文本框背景填充幻灯片背景格式设置添加考生文件夹下的版式 NO12345 插入幻灯片和放入图片☞快速:插入→相册→新建相册→文件→图片版式→相框形状→调整边框宽度左下角背景图片:视图→…...
Baumer工业相机堡盟相机的相机传感器芯片清洁指南
Baumer工业相机堡盟相机的相机传感器芯片清洁指南 Baumer工业相机1.Baumer工业相机传感器芯片清洁工具和清洁剂2.Baumer工业相机传感器芯片清洁步骤2.1、准备步骤2.2、清洁过程1.定位清洁工具2.清洁传感器3.使用吹风装置 Baumer工业相机传感器芯片清洁的优势设计与结…...
Spring Boot 整合 JPA 实现数据持久化
目录 前言 一、JPA 核心概念与实体映射 1. 什么是 JPA? 2. JPA 的主要组件 3. 实体映射 4. 常见的字段映射策略 二、Repository 接口与自定义查询 1. 什么是 Repository 接口? 2. 动态查询方法 3. 自定义查询 4. 分页与排序 三、实战案例&…...
快速在wsl上部署学习使用c++轻量化服务器-学习笔记
知乎上推荐的Tinywebserver这个服务器,快速部署搭建,学习c服务器开发 仓库地址 githubhttps://link.zhihu.com/?targethttps%3A//github.com/qinguoyi/TinyWebServerhttps://link.zhihu.com/?targethttps%3A//github.com/qinguoyi/TinyWebServer 在…...
【R语言】数据操作
一、查看和编辑数据 1、查看数据 直接打印到控制台 x <- data.frame(a1:20, b21:30) x View()函数 此函数可以将数据以电子表格的形式进行展示。 用reshape2包中的tips进行举例: library("reshape2") View(tips) head()函数 查看前几行数据&…...
MariaDB MaxScale实现mysql8主从同步读写分离
一、MaxScale基本介绍 MaxScale是maridb开发的一个mysql数据中间件,其配置简单,能够实现读写分离,并且可以根据主从状态实现写库的自动切换,对多个从服务器能实现负载均衡。 二、MaxScale实验环境 中间件192.168.121.51MaxScale…...
【python】简单的flask做页面。一组字母组成的所有单词。这里的输入是一组字母,而输出是所有可能得字母组成的单词列表
目录结构如下: https://github.com/kaede316/Pythons_pj.git 效果: 后续可扩展为工具网站: 更新 2025.02.09 1、增加等间距制作人 时间信息 2、增加判断润年的功能...
单片机之基本元器件的工作原理
一、二极管 二极管的工作原理 二极管是一种由P型半导体和N型半导体结合形成的PN结器件,具有单向导电性。 1. PN结形成 P型半导体:掺入三价元素,形成空穴作为多数载流子。N型半导体:掺入五价元素,形成自由电子作为多…...
吴恩达深度学习——卷积神经网络的特殊应用
内容来自https://www.bilibili.com/video/BV1FT4y1E74V,仅为本人学习使用。 文章目录 人脸识别相关定义Similarity函数使用Siamese网络实现函数d使用Triplet损失学习参数 神经风格迁移深度卷积网络可视化神经风格迁移的代价函数内容损失函数风格损失函数 人脸识别 …...
安宝特方案 | AR助力制造业安全巡检智能化革命!
引言: 在制造业中,传统巡检常面临流程繁琐、质量波动、数据难以追溯等问题。安宝特AR工作流程标准化解决方案,通过增强现实AR技术,重塑制造业安全巡检模式,以标准化作业流程为核心,全面提升效率、质量与…...
Unity-Mirror网络框架-从入门到精通之Discovery示例
文章目录 前言Discovery示例NetworkDiscoveryNetworkDiscoveryHUDServerRequestServerResponse最后前言 在现代游戏开发中,网络功能日益成为提升游戏体验的关键组成部分。本系列文章将为读者提供对Mirror网络框架的深入了解,涵盖从基础到高级的多个主题。Mirror是一个用于Un…...
项目的虚拟环境的搭建与pytorch依赖的下载
文章目录 配置环境 pytorch的使用需要安装对应的cuda 在PyTorch中使用CUDA, pytorch与cuda不同版本对应安装指南,查看CUDA版本,安装对应版本pytorch 【超详细教程】2024最新Pytorch安装教程(同时讲解安装CPU和GPU版本) 配置环境…...
Chapter03-Authentication vulnerabilities
文章目录 1. 身份验证简介1.1 What is authentication1.2 difference between authentication and authorization1.3 身份验证机制失效的原因1.4 身份验证机制失效的影响 2. 基于登录功能的漏洞2.1 密码爆破2.2 用户名枚举2.3 有缺陷的暴力破解防护2.3.1 如果用户登录尝试失败次…...
vscode(仍待补充)
写于2025 6.9 主包将加入vscode这个更权威的圈子 vscode的基本使用 侧边栏 vscode还能连接ssh? debug时使用的launch文件 1.task.json {"tasks": [{"type": "cppbuild","label": "C/C: gcc.exe 生成活动文件"…...
解锁数据库简洁之道:FastAPI与SQLModel实战指南
在构建现代Web应用程序时,与数据库的交互无疑是核心环节。虽然传统的数据库操作方式(如直接编写SQL语句与psycopg2交互)赋予了我们精细的控制权,但在面对日益复杂的业务逻辑和快速迭代的需求时,这种方式的开发效率和可…...
376. Wiggle Subsequence
376. Wiggle Subsequence 代码 class Solution { public:int wiggleMaxLength(vector<int>& nums) {int n nums.size();int res 1;int prediff 0;int curdiff 0;for(int i 0;i < n-1;i){curdiff nums[i1] - nums[i];if( (prediff > 0 && curdif…...
macOS多出来了:Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用
文章目录 问题现象问题原因解决办法 问题现象 macOS启动台(Launchpad)多出来了:Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用。 问题原因 很明显,都是Google家的办公全家桶。这些应用并不是通过独立安装的…...
第一篇:Agent2Agent (A2A) 协议——协作式人工智能的黎明
AI 领域的快速发展正在催生一个新时代,智能代理(agents)不再是孤立的个体,而是能够像一个数字团队一样协作。然而,当前 AI 生态系统的碎片化阻碍了这一愿景的实现,导致了“AI 巴别塔问题”——不同代理之间…...
现代密码学 | 椭圆曲线密码学—附py代码
Elliptic Curve Cryptography 椭圆曲线密码学(ECC)是一种基于有限域上椭圆曲线数学特性的公钥加密技术。其核心原理涉及椭圆曲线的代数性质、离散对数问题以及有限域上的运算。 椭圆曲线密码学是多种数字签名算法的基础,例如椭圆曲线数字签…...
【Zephyr 系列 10】实战项目:打造一个蓝牙传感器终端 + 网关系统(完整架构与全栈实现)
🧠关键词:Zephyr、BLE、终端、网关、广播、连接、传感器、数据采集、低功耗、系统集成 📌目标读者:希望基于 Zephyr 构建 BLE 系统架构、实现终端与网关协作、具备产品交付能力的开发者 📊篇幅字数:约 5200 字 ✨ 项目总览 在物联网实际项目中,**“终端 + 网关”**是…...
Caliper 配置文件解析:config.yaml
Caliper 是一个区块链性能基准测试工具,用于评估不同区块链平台的性能。下面我将详细解释你提供的 fisco-bcos.json 文件结构,并说明它与 config.yaml 文件的关系。 fisco-bcos.json 文件解析 这个文件是针对 FISCO-BCOS 区块链网络的 Caliper 配置文件,主要包含以下几个部…...
智能分布式爬虫的数据处理流水线优化:基于深度强化学习的数据质量控制
在数字化浪潮席卷全球的今天,数据已成为企业和研究机构的核心资产。智能分布式爬虫作为高效的数据采集工具,在大规模数据获取中发挥着关键作用。然而,传统的数据处理流水线在面对复杂多变的网络环境和海量异构数据时,常出现数据质…...
