Android环境下 MQTT+Protocol Buffers实现消息推送

曾记否，大学初识Android时，做出一款聊天软件曾是多少少年的梦想。。。（好吧可能只是我的)

emm… 怎么写？长连接，那就直接怼socket吧！于是闷起脑壳就开始写。服务端+客户端，一通操作猛如虎，并伴随着各种线程异常之后终于是肝出来一个聊天室，，，好像还阔以，但是随着少年长大，你渐渐明白服务的重要性，这样的连接你可能保持不了几个，想实现消息推送,MQTT了解一下？

MQTT(消息队列遥测传输)

一个基于客户端-服务器的消息发布/订阅传输协议。优点：轻量、简单、开放、易于实现;由于其低开销、低带宽占用，所以在物联网、小型设备、移动应用等方面有广泛的应用
MQTT协议有三种身份： 发布者、代理、订阅者，发布者和订阅者都为客户端，代理为服务器，同时消息的发布者也可以是订阅者
MQTT传输的消息分为主题（Topic，可理解为消息的类型，订阅者订阅后，就会收到该主题的消息内容payload。
运行流程:

特点如下

使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序之间的耦合。

对负载内容屏蔽的消息传输。

使用 TCP/IP 提供基础网络连接。

小型传输，开销很小(固定长度的是头部是2个字节)，协议交换最小化，以降低网络流量
整体上协议可拆分为：固定头部+可变头部+消息体

提供一种机制，使得客户端异常中断时，能够使用LastWill和Testament特性通知有关各方
Last Will：即遗言机制，用于通知同一主题下的其他设备发送遗言的设备已经断开了连接。
Testament：遗嘱机制，功能类似于Last Will。

有三种级别消息发布服务质量：
qos为0：“至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复，不久会发送第二次。
qos为1：“至少一次”，确保消息到达，但消息重复可能会发生。
qos为2：“只有一次”，确保消息到达一次。

有了`MQTT`代替`Socket`作为传输协议，但是数据结构呢？你也许会说`json`，但在通讯极其频繁的IM或者推送泛滥的场景中，`Protocol Buffers`更加适用（当然各有各的优缺点，比如`json`易读，`Protocol Buffers`**小**、**快**），当然使用它我们得有后端同学支持

有了协议，我们还需要一个服务来为我们提供消息的订阅发送，我使用的是EMQ，需要后端同学将自己的服务与EMQ服务接入进行通讯

##EMQ

EMQ X Broker 是基于高并发的 Erlang/OTP 语言平台开发，支持百万级连接和分布式集群架构，发布订阅模式的开源 MQTT 消息服务器。
EMQ X Broker 在全球物联网市场广泛应用。无论是产品原型设计、物联网创业公司、还是大规模的商业部署，EMQ X Broker 都支持开源免费使用。
详情：https://www.emqx.io/cn/

##Protocol Buffer

Protocol Buffers 是Google推出的用于序列化结构化数据的灵活、高效、自动化的机制。它与 XML、JSON一样都是结构数据序列化的工具，但ProtoBuffer更小，更快，更简单。开发者需要定义一次构造数据的方式，然后就可以使用特定代码轻松地在各种语言的各种数据流中写入和读取数据。同样，使用它必须有后端同学的支持
详情：https://developers.google.com/protocol-buffers

优点

性能较xml,json, thirft等好，效率高
代码生成机制，数据解析类自动生成
支持向后兼容和向前兼容
支持多种编程语言（java，c++，python）

缺点

二进制格式导致可读性差
缺乏自描述

先上个 Protocol Buffers 与 json 的对比图

//json：
{
    "name":"Fenrir",
    "age":6
    "age1":6
    "age2":6
    "age3":6
    "age4":6
    "age5":6
    "age6":6
    "age7":6
    "age8":6
    "age9":6
    "age10":6
    "age11":6
}

//Protocol Buffers:
10 6 70 101 110 114 105 114 16 34 24 46 32 56 40 16 48 56 56 86 64 66 72 56 80 86 88 54 96 66 104 66

//或者将其Base64，变得下面这样
CgZGZW5yaXIQIhguIDgoEDA4OFZAQkg4UFZYNmBCaEI=

QA：？？？

`Protocol Buffers`将数据从对象转换成了一个`byte数组`。诚然，由于其可读性差，测试工作会变得困难;
    但是使用`Protocol Buffers`不仅能降低数据传输代价，并且无论是在客户端还是服务端，对于数据的`序列化`与`反序列化`速度都将得到不小的提升；所以这并不应该成为使用它的障碍，在真正实装起来时就需要我们给QA们写一个序列化与反序列化的工具。

##在Android使用Protocol Buffers
project gradle中

dependencies {
    classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.8'
}

app gradle中

apply plugin: 'com.google.protobuf' //应用proto插件
    protobuf {
        protoc {
            artifact = 'com.google.protobuf:protoc:3.0.0'
        }
        plugins {
            javalite {
                artifact = 'com.google.protobuf:protoc-gen-javalite:3.0.0'
            }
        }
        generateProtoTasks {
            all().each { task ->
                task.builtins {
                    remove java
                }
                task.builtins {
                    java {}
                    cpp {}
                }
            }
        }
    }
    sourceSets {
        main {
            java {
                srcDir 'src/main/java'
            }
            proto {
                srcDir 'src/main/proto' //配置proto文件路径
            }
            jniLibs.srcDir 'libs'
            jni.srcDirs = []    //disable automatic ndk-build
        }
    }
dependencies {
    api 'com.google.protobuf:protobuf-java:3.6.1'
    api 'com.google.protobuf:protoc:3.10.1'
    implementation 'com.squareup.retrofit2:converter-protobuf:2.2.0'
}

这里以构建一个无限发送心跳包的程序为例，这里实现心跳包的目的如下

为服务端提供信息用于判断用户是否下线
确保与服务端建立的topic的唯一性
顺便增加一下Android服务的优先级

使用ProtocolBuffers前需要先编辑.proto文件，语法过多就不写在这里了，请移步至https://developers.google.com/protocol-buffers/docs/proto3
在src/main下创建文件夹proto并在其目录下创建文件 MqttHeartBeatMessage.proto

syntax = "proto3"; //protoBuf版本
package rubbishcommunity; //包
option java_package = "com.example.xxxxxxx"; //包名
option java_outer_classname = "MqttHeartBeatMessageOutClass"; //生成的Java Class文件类名

message MqttHeartBeatMessage{
    int64 uin = 1;
    string linkKey = 2;
    int64 timestamp = 3;
}

.proto文件编辑完成之后 Rebuild 编译一下项目,可以发现在java(generated)下面已经生成了proto编译成的java文件MqttHeartBeatMessageOutClass.java，里面包含了MqttHeartBeatMessage.java并实现了get、set方法；

到此，Proto部分算是完成了，接下来我们需要使用MQTTService实现无限发送心跳包,但在这之前，你要先了解下面两位

##Eclipse Paho

Eclipse提供的一个访问MQTT服务器的一种开源客户端库
https://www.eclipse.org/paho/clients/android/

##MqttAndroidClient

简单的说就是对MQTTService的连接、通讯等方法的封装类
[https://www.eclipse.org/paho/files/android-javadoc/org/eclipse/paho/android/service/MqttAndroidClient.html]

引入MQTT的Service以及对Android相关资源

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

Manifests中注册MqttService的Service

<service android:name="org.eclipse.paho.android.service.MqttService"/>

在项目中使用MqttAndroidClient进行MQTT连接与通讯
理论上应该将MqttAndroidClient的生命周期与Application绑定在一起，在我的毕设项目中我是再写了一个Service，在我的MainActivity打开之后开启，并在Service中轮询发送心跳包；

初始化代码：

//新建Client,以设备ID作为client ID
mqttAndroidClient = MqttAndroidClient(
	context,
	MQTT_URL, //MQTT服务地址
	CLIRNT_ID //MQTT连接在代理服务处的clientId，一般由后端同学定义
)
mqttAndroidClient?.setCallback(object : MqttCallbackExtended {
	override fun connectComplete(reconnect: Boolean, serverURI: String) {
		//连接成功
		if (reconnect) {
			// 由于clean Session ，我们需要重新订阅
			try {
				subscribeToTopic()
			} catch (e: Exception) {
			    ex.printStackTrace()
			}
		}
	}
	
	override fun connectionLost(cause: Throwable) {
		//连接断开
	}
	
	override fun messageArrived(topic: String, message: MqttMessage) {
              //接收到订阅的消息
	}
	
	override fun deliveryComplete(token: IMqttDeliveryToken) {
		//服务器成功delivery消息
	}
})
//新建连接设置
val mqttConnectOptions = MqttConnectOptions()
//断开后，是否自动连接
mqttConnectOptions.isAutomaticReconnect = true
//是否清空客户端的连接记录。若为true，则断开后，broker将自动清除该客户端连接信息
mqttConnectOptions.isCleanSession = true
//设置Mq连接的userName
mqttConnectOptions.userName = getLocalEmail()
//设置超时时间，单位为秒
//mqttConnectOptions.setConnectionTimeout(2);
//心跳时间，单位为秒。即多长时间确认一次Client端是否在线（此心跳时间非刚才说的心跳时间，这个是MqttService内部的心跳）
mqttConnectOptions.keepAliveInterval = 2
//允许同时发送几条消息（未收到broker确认信息）
mqttConnectOptions.maxInflight = 2
//选择MQTT版本
mqttConnectOptions.mqttVersion = MqttConnectOptions.MQTT_VERSION_3_1_1
try {
	//开始连接
	mqttAndroidClient?.connect(mqttConnectOptions, this, object : IMqttActionListener {
		override fun onSuccess(asyncActionToken: IMqttToken) {
			val disconnectedBufferOptions = DisconnectedBufferOptions()
			disconnectedBufferOptions.isBufferEnabled = true
			disconnectedBufferOptions.bufferSize = 100
			disconnectedBufferOptions.isPersistBuffer = false
			disconnectedBufferOptions.isDeleteOldestMessages = true
			mqttAndroidClient!!.setBufferOpts(disconnectedBufferOptions)
			subscribeToTopic()//成功连接以后开始订阅
		}
		
		override fun onFailure(asyncActionToken: IMqttToken, exception: Throwable) {
			//连接失败
			exception.printStackTrace()
		}
	})
} catch (ex: MqttException) {
	ex.printStackTrace()
}

轮询发送心跳包

Observable.interval(1, 3, TimeUnit.SECONDS)
	.doOnNext {
		mqService //这是我的服务，没有的话直接使用MqAndroidClient就好
			.publishMessage(
				Base64.encodeToString(//下面就是构造proto生成的Class的对象了
					MqttHeartBeatMessageOutClass.MqttHeartBeatMessage.newBuilder()
						.setLinkKey(getLinkKey()) //proto中自定义的属性
						.setUin(getLocalUin()) //proto中自定义的属性
						.setTimestamp(System.currentTimeMillis()) //proto中自定义的属性
						.build() //构造结束
                              .toByteArray(),
					NO_WRAP
				),
				DEW_MQTT_HEART_BEAT_TOPIC //心跳包的TOPIC
			)
	}.subscribe()

至此，心跳包发送已经完成了，接下来说说接收Protocol Buffer数据

在上面代码中应该看见了，在MqttAndroidClient接收的数据在messageArrived(topic: String, message: MqttMessage)方法中，实际上这里并非所有的消息都会是protocol Buffers消息，所以这里我们可以想办法区分，这里就不累述了。要说的是Protocol Buffer的消息在内部也是需要区分的，比如有两种消息过来1⃣️我的朋友圈有人点赞我需要知道是谁点赞2⃣️有人给我发消息我需要知道是什么消息，而这两种消息内部返回的数据结构可能就完全不一样，所以需要用到Protocol Buffers的enum进行区分，那么我们与后端同学一起定义一种协议,也就是接收的ProtocolBuffers消息的数据结构，像下面这样：

syntax = "proto3";
package rubbishcommunity;
option java_package = "com.example.rubbishcommunity";
option java_outer_classname = "NotifyMessageOutClass";

enum NotifyType{
    ERROR_DEFAULT_TYPE = 0; //默认
    SYNC_NEW_MESSAGE = 1; //新消息
    SYNC_MOMENTS_COMMENT = 2; //有人点赞
}

message NotifyMessage{
    Header header = 1; //数据头
    string payload = 2; //真实数据（虽然是String，实际是后端将Protocol Buffers数据Base64之后的String）
}

message Header{
    NotifyType notifyType = 1; //消息类型
    int64 timestamp = 2; //时间戳
    string notifyUUID = 3; //消息UUID
}

之后我们就可以愉快地解析数据啦！举个栗子

MqttAndroidClient初始化的地方：

             //订阅的消息送达
override fun messageArrived(topic: String, message: MqttMessage) {

	//先用Base64解码
	val resBytes = Base64.decode(message.payload, Base64.NO_WRAP)
	
	//得到最外层的NotifyMessage对象
	val notifyMessage = NotifyMessageOutClass.NotifyMessage.parseFrom(resBytes)
	
	//发送MQ事件，目前在BasegFragment中有处理事件的方法
	sendMQData(MQNotifyData(notifyMessage.header.notifyType,notifyMessage.payload))
	
}

BaseFragment中：

//实际'MQ消息'处理者
private fun handleMQNotifyMessage() {
	notifyDisposable = getMQNotifyObs()
		.observeOn(AndroidSchedulers.mainThread())
		.doOnNext {
				onMQMessageArrived(it)
		}.subscribe({}, { Timber.e(it) })
}

实现的Fragment中：

//有MQ消息
override fun onMQMessageArrived(mqNotifyData: MQNotifyData) {
	when (mqNotifyData.mqNotifyType) {
		NotifyMessageOutClass.NotifyType.SYNC_MOMENTS_FAVORITE -> { //有人点赞
			//弹个通知并更新列表对应Item
		}
		else -> {
		
		}
	}
}

将得到的数据包装成自定义的一个MQNotifyData数据类，数据类中有header和payload，再通过sendMQData()发送到Fragment或者Activity中去，我是在Fragment中一直有一个mq消息的Subject流，当sendMQData()就触发onNext()，其中调用BaseFragment的方法，具体实现在具体需要MQ消息的Fragment中。这是我的方法而已，就是将数据通知到界面，如何实现随意，欢迎大佬来指正～

MQ中的ProtocolBuffers完成了，还有最后一步就是为我们的正常Http请求的数据也添加支持，啊好像写不完了，各位看官～下次补上吧～！！

网络通讯

本博客所有文章除特别声明外，均采用 CC BY-SA 4.0 协议，转载请注明出处！

Android View 从测量布局到触摸反馈上一篇

中间凹陷的BottomNavigationView（仿百度地图）下一篇