Android基于MQTT的长连接实现：连接、订阅、心跳与自动重连

简介面向Android开发者的MQTT长连接示例工程围绕发布/订阅模型展示Paho客户端的集成方式实现消息实时收发、主题订阅、断线自动重连、离线消息补偿及SSL/TLS安全连接等核心机制适合有即时通信、消息推送或IoT控制需求的移动端开发者。压缩包含1278个文件约21.23MB以xml布局、png图片、class编译产物、jar依赖、java源码及json配置为主并附带可直接安装的app-debug.apk、AIDL接口和gradle构建文件便于对照工程结构理解运行流程。目前已有1217人学习下载demo对连接失败重试、QoS等级选择、异常资源释放等细节均有覆盖可直接迁移到实际项目中作为长连接模块的基础。1. 为什么要用MQTT做Android长连接先聊一个很多做过即时通讯或者消息推送的Android开发都会遇到的痛点轮询太费电、太费流量而且做不到真正的实时自己拿Socket手写一个长连接又要自己处理心跳、粘包拆包、断线重连、消息去重一套搞下来少说一两周还容易埋一堆坑。MQTTMessage Queuing Telemetry Transport就是为这种场景设计的轻量级消息传输协议底层基于TCP特点是极简、省流量、支持QoS消息质量分级、支持服务端主动推送。它最早用在卫星通信和物联网传感器上但放在移动端做长连接消息通道同样非常合适很多App的推送、IM、设备控制都跑在MQTT上。这次要拆解的项目就是一个最小的Android端MQTT长连接Demo麻雀虽小但五脏俱全包含连接、订阅、发布、收消息、心跳保活、断线自动重连这几个核心能力。不管你是想做一个物联网控制端、消息推送通道还是想给现有App补一个稳定的长连接能力这个Demo都可以作为起步骨架照着改改就能用。我用的这套方案是Eclipse Paho Android这也是目前Android端最主流的MQTT客户端组合。Paho是Eclipse基金会维护的开源MQTT客户端库Android版本封装好了Service、BroadcastReceiver等组件省去了自己管理多线程和进程通信的麻烦。2. 协议基础理解MQTT的发布订阅模型和几个关键点2.1 不是点对点而是发布/订阅MQTT和HTTP最大的区别HTTP是请求-响应客户端主动拉MQTT是发布/订阅客户端订阅某个主题Topic只要有人向这个主题发布消息所有订阅者都能收到。打个比方HTTP就像你打电话给某个人问情况得对方接了你才有信息MQTT就像你订了一份报纸邮局Broker负责分发只要报纸印出来订了的人都会收到不用你反复打电话去问。正因为这个模型MQTT特别适合一对多的消息分发场景比如设备状态广播、群聊消息、行情推送。一台车机、一块屏幕、一个手环只要有MQTT客户端都能订阅同一个Topic实时收数据。2.2 Broker、Topic、QoS三件套Broker消息的中转站可以理解成邮局。Android客户端不直接通信都连到同一个Broker上。常见的Broker有开源的EMQX、Mosquitto以及云厂商提供的MQTT服务。Topic消息的分类标签用斜杠分层比如device/{deviceId}/status支持通配符匹配单层#匹配多层。QoS消息投递的可靠性等级。QoS 0最多一次消息可能丢失适合环境传感器数据这种丢了无所谓的场景。QoS 1至少一次保证到达但可能重复需要业务侧做去重。QoS 2恰好一次性能开销最大适合命令下发、支付通知这类不能丢也不能重的场景。2.3 为什么心跳和保活是长连接的命根子TCP长连接有个天然问题网络中间设备比如路由器、运营商网关会清理空闲连接。如果一个连接长时间没有数据流动中间设备就认为它已经死了直接把这个连接从NAT表里踢掉。之后数据再发过来就会石沉大海。MQTT的心跳机制就是用来解决这个问题的。客户端每隔一段固定时间Keep Alive间隔发一个PINGREQ报文Broker收到后回PINGRESP。这样既保持了连接活跃又能让客户端感知到连接是否还通着。Keep Alive间隔不是越大越好也不是越小越好。太大会被中间设备清理掉太小会频繁唤醒网络模块增加耗电和流量。一般移动网络建议90秒到120秒实际项目中通常还会配一个30秒到60秒的保活小包空Topic或者心跳Topic双保险。3. 工程搭建引入Paho库和配置环境3.1 依赖引入在build.gradleModule级别的dependencies里加上implementation org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5 implementation org.eclipse.paho:org.eclipse.paho.android.service:1.1.1注意android.service这个包是专门给Android用的封装了后台Service没有它的话你得自己在线程池里维护连接状态复杂度会高不少。如果你的项目用的是AndroidX还需要额外加一个兼容包implementation androidx.localbroadcastmanager:localbroadcastmanager:1.1.0Paho的Android Service内部用的是本地广播来通知消息到达不在AndroidX项目里加上这个依赖运行时会直接崩。3.2 配置AndroidManifest权限MQTT连接需要网络权限同时为了防止屏幕关闭后CPU休眠导致连接断开还需要WAKE_LOCK权限uses-permission android:nameandroid.permission.INTERNET / uses-permission android:nameandroid.permission.WAKE_LOCK / uses-permission android:nameandroid.permission.ACCESS_NETWORK_STATE /另外要在application节点里注册Paho的核心Serviceservice android:nameorg.eclipse.paho.android.service.MqttService /这个Service是长连接常驻的关键。它会在后台维护所有的MQTT连接实例只要进程不死、连接不断就能持续收消息。3.3 选一台能用的Broker自己做开发调试的话我推荐用Docker直接起一个本地EMQX一分钟搞定docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 18083:18083 emqx/emqx:5.0启动后1883端口是MQTT TCP接入端口18083端口是Dashboard管理后台浏览器打开http://localhost:18083默认账号admin密码public8083端口是WebSocket接入端口给Web端联调用没有Docker环境的话也可以直接连EMQX官方的公共Brokerbroker.emqx.io端口1883无需账号密码开发调试完全够用。4. Demo核心实现连接、订阅、发布、重连一气呵成4.1 创建MQTT客户端实例Paho Android版本推荐用到的类是MqttAndroidClient它跟Java标准版的MqttClient最大的不同是绑定了Android的Context并运行在Service进程中可以自动处理断网后的重新连接。private MqttAndroidClient mqttClient; private void initMqttClient() { String serverUri tcp://broker.emqx.io:1883; String clientId android_demo_ System.currentTimeMillis(); mqttClient new MqttAndroidClient(context, serverUri, clientId); mqttClient.setCallback(new MqttCallbackExtended() { Override public void connectComplete(boolean reconnect, String serverURI) { // 连接成功包括自动重连成功在这里重新订阅Topic if (reconnect) { subscribeTopic(device/status); } } Override public void connectionLost(Throwable cause) { // 连接丢失Paho内部会自动尝试重连 } Override public void messageArrived(String topic, MqttMessage message) { // 收到订阅的消息 String payload new String(message.getPayload()); Log.d(MQTT, topic: topic , message: payload); } Override public void deliveryComplete(IMqttDeliveryToken token) { // 消息发布成功 } }); }这里有个非常容易踩的坑clientId在同一个Broker下必须唯一。如果你用同一个clientId两头连后连的那个会把先连的挤下线Broker会强制断开前一个连接。我之前调试时用固定ID连着连连断查了半天才发现是IDE的模拟器和真机同时连了同一个ID。4.2 配置连接参数并建立连接MqttConnectOptions这个类就是连接参数的集合核心配置有这几个private void connect() { MqttConnectOptions options new MqttConnectOptions(); // 是否清空Sessionfalse表示保留会话状态断线重连后可以接收离线消息 options.setCleanSession(false); // 心跳间隔单位秒一般60~120 options.setKeepAliveInterval(60); // 设置超时时间 options.setConnectionTimeout(10); // 自动重连这个必须开 options.setAutomaticReconnect(true); // 遗嘱消息客户端异常掉线时由Broker代替发布这条消息 options.setWill(device/will, offline.getBytes(), 1, true); try { mqttClient.connect(options); } catch (MqttException e) { e.printStackTrace(); } }几个参数单独说一下setCleanSession(false)意思是Broker会为这个客户端保留会话状态包括订阅关系和离线期间的消息。这样客户端掉线重连后不需要重新订阅也能收到离线期间的QoS 1/2消息。代价是Broker需要额外维护Session状态占用一定资源。如果你只是做实时消息推送、不需要收离线消息用true反而更省。setWill()遗嘱消息是MQTT一个很有特色的机制。正常退出时客户端可以主动发DISCONNECT报文Broker知道你是正常下线的但如果客户端是直接断网、崩溃Broker会在检测到连接异常后代为向遗嘱Topic发一条消息。这在设备在线状态管理里非常有用配合LWT机制可以实现设备上下线的准确感知。setAutomaticReconnect(true)这个参数开之后Paho会在连接意外断开后自动按退避策略重连不需要自己写循环重连的逻辑省了很多事。还有一个需要重点说的问题connect()是异步的。它内部会启动一个后台任务建立TCP连接所以在调用connect()之后立刻判断连接是否成功是拿不到结果的。正确做法是在MqttCallbackExtended的connectComplete回调里做后续操作比如订阅Topic。4.3 订阅和发布 Topic订阅private void subscribeTopic(String topic) { try { // 参数Topic、QoS等级 mqttClient.subscribe(topic, 1); } catch (MqttException e) { e.printStackTrace(); } }发布消息private void publishMessage(String topic, String content) { try { MqttMessage message new MqttMessage(); message.setPayload(content.getBytes()); // QoS等级1保证消息至少送达一次 message.setQos(1); message.setRetained(true); mqttClient.publish(topic, message); } catch (MqttException e) { e.printStackTrace(); } }setRetained(true)这个参数值得展开说一下。保留消息的意思是Broker会为你发布的这条消息存一个最新的副本新的订阅者订阅这个Topic时会立刻收到一条这个副本消息而不是需要等下一个发布者发新消息。这个特性太适合做设备状态同步了。比如设备A上线后发布一条online的保留消息设备B隔了很久才订阅这个Topic它也能立刻收到online这个状态不用等设备A再发一次。很多智能家居的控制端就是这个套路。QoS等级选多少要结合场景普通控制指令建议QoS 1保证到达且性能开销能接受系统内部高频心跳、遥测数据用QoS 0丢几条无所谓计费、支付、开锁这类关键指令才用QoS 24.4 断开连接的正确姿势在Activity销毁或者不需要长连接时一定要显式断开否则Service还挂在后台浪费电也占着连接private void disconnect() { try { mqttClient.disconnect(); } catch (MqttException e) { e.printStackTrace(); } }这里要注意一个顺序问题先disconnect()再close()。只调close()也有隐含的disconnect操作但某些Paho版本里直接close()不会发送DISCONNECT报文Broker只能靠心跳超时判定你掉线中间会有一段假在线的时间窗口。我在项目里还遇到过一种情况在onDestroy里调了disconnect()后立马退出App但Paho的断开是异步的DISCONNECT报文还没发出去进程就死了Broker还是判定异常掉线遗嘱消息被触发。解决办法很简单disconnect()之后加一个非常短的Thread.sleep(100)让报文有时间发出去。虽然这个做法不太优雅但在实际项目中确实有效。5. 长连接可靠性的几个关键细节5.1 网络切换时的自动恢复手机上最常见的场景是WiFi切到4G/5G或者从地下室出来信号恢复。这些场景下TCP连接基本都会断开Paho的AutomaticReconnect会自动触发重连。但有个细节Paho默认的重连退避策略是1秒、2秒、4秒...最大到32秒然后一直循环。这个策略对移动端来说偏激进频繁唤醒网络模块会额外耗电。如果项目对电量敏感建议自己接管重连逻辑关闭自动重连用一个带退避算法的工作线程来管理。另外我强烈建议在项目里监听系统网络状态变化。网络恢复时可以主动触发一次连接检查// 需要一个网络状态监听器注册到ConnectivityManager private void onNetworkAvailable() { if (mqttClient ! null !mqttClient.isConnected()) { connect(); } }这样可以在系统恢复网络后立刻重连不用等Paho的下一次心跳超时发现对用户体验有明显帮助。5.2 进程被杀死后的恢复方案Android系统会在内存不足时杀掉后台进程这是长连接最大的敌人。Paho的Service能在一定程度上提高进程优先级但不能保证不被杀。一个通用的做法是使用前台Service。在Android上给Service设置一个常驻通知比如后台运行中把进程优先级提到前台级别被系统清理的概率会大幅降低Notification notification createForegroundNotification(); startForeground(1, notification);如果是做IoT类的设备控制App还有一种更狠的方案用双进程守护一个主进程一个守护进程互相拉起。不过这个方案实施起来复杂而且在高版本Android上限制越来越严不建议新手搞前台Service已经能覆盖绝大多数场景了。从Android 12API 31开始前台Service启动限制增多如果你的App只在App处于前台时才需要收消息其实不用常驻Service在onResume里连、onPause里断即可。5.3 消息收发缓冲区与消费速度Paho客户端内部有一个默认的消息分发线程。如果业务处理速度跟不上消息到达速度消息就会在内存里堆积。某些极端场景下大量积压消息会导致内存飙升甚至OOM。建议在messageArrived回调里不要做耗时操作只做轻量的数据解析和UI转发。如果业务逻辑重通过Handler切到工作线程或者用LiveData/RxJava把消息发射出去保证回调线程尽快返回。5.4 开发期调试的三个实用技巧调试长连接经常会遇到消息没收到或连接一会被踢这类问题我总结了一套排查顺序先看Broker Dashboard的客户端列表。能显示当前有哪些客户端在连接用的是哪个ClientID有没有被踢下线的记录。比如EMQX管理后台的客户端页面就能看到每个连接的在线状态消息流向也能在消息追踪里看到。在本地起一个命令行MQTT客户端订阅同一个Topic做对照测试。如果命令行能收到消息而App收不到问题在App侧如果命令行也收不到问题在Broker侧或者发布端。打开Paho的日志开关。Paho使用了标准的java.util.logging在Android上可以和Log桥接或者直接用adb logcat看系统日志。有个专门用来测试MQTT的桌面客户端叫MQTTX跨平台图形化界面支持订阅、发布、收发历史查看调试时特别好用推荐给所有做MQTT开发的人。6. 常见问题速查表实际操作中下面这些问题都是高频出现的我整理了一张排查表照着一个个排除比自己瞎猜快得多现象可能原因解决办法connect()一直不回调Broker地址不通、端口错误、防火墙拦截先用telnet测端口telnet broker.emqx.io 1883确认Broker是否启用了TLS但客户端用了tcp://连接成功但收不到消息订阅的Topic不对、QoS配置错误、订阅操作晚于消息发布检查Topic精确拼写用MQTTX做一个对照订阅确认是否设置了Retained需要主动拉取经常掉线KeepAlive间隔过长网络切换频繁路由器清理空闲连接缩小setKeepAliveInterval到60秒左右开启自动重连考虑增加保活小包一个客户端一上线另一个就被踢掉ClientID重复确保每台设备、每个安装实例的ClientID唯一建议用UUID离线状态下发的消息收不到setCleanSession(true)导致无离线消息QoS为0不存离线消息使用setCleanSession(false)发送时使用QoS 1或2确认保留消息语义是否符合需求电量消耗明显增加心跳频繁、Service常驻、频繁重连调整心跳间隔到合理范围减少重连频率考虑在后台时降低心跳频率App被杀后收不到消息进程被系统回收使用前台Service在应用内加大setAutomaticReconnect的重试窗口补充一个很难排查的场景Android的Doze模式。屏幕关闭且设备静止一段时间后系统会进入Doze模式限制网络访问和唤醒。如果设备在Doze模式下长时间收不到消息不是MQTT出了问题而是系统限制了后台网络。解决方法是向系统申请忽略电池优化白名单或者在Doze模式下用高优先级FCM通道唤醒App。7. Demo后续可以怎么扩展这个Demo跑通之后顺着这个骨架往下走有几个很自然的方向第一是加TLS加密。生产环境不建议明文传输1883端口换8883配置CA证书几行代码的事安全性提升一大截。第二是消息体的业务化。现在Demo里收的是纯字符串后面可以定义自己的协议结构比如用JSON封装消息类型、序列号、时间戳在messageArrived里做统一反序列化和分发。第三是离线消息策略。如果场景要求App离线期间很多关键消息不能丢可以配合setCleanSession(false)同时给Topic定制合适的QoS等级让Broker代为缓存。第四是做多Topic的应用层路由。比如device/{deviceId}/command、device/{deviceId}/telemetry、device/{deviceId}/event按业务域拆成多个Topic消息管理更清晰订阅端也能灵活按需订阅。我在实际使用中遇到过很多长连接的细节问题比如心跳冲突、Socket超时时间设置不合理、重连风暴压垮Broker等等。这些问题说大不大但排查起来非常耗时。如果你也正在做一个基于MQTT的Android长连接项目建议先把这个Demo彻底吃透跑通连接、订阅、发布、断网重连这一整套流程再去碰那些复杂的生产级需求。把这套基础打扎实了后面遇到问题你就能快速定位到具体环节不用再一头雾水地到处查资料了。本文还有配套的精品资源点击获取