MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议。它是一种发布/订阅，极其简单和轻量级的消息传递协议，专为受限设备和低带宽，高延迟或不可靠的网络而设计。它的设计思想是轻巧、开放、简单、规范，易于实现。这些特点使得它对很多场景来说都是很好的选择，特别是对于受限的环境如机器与机器的通信（M2M）以及物联网环境。相对于XMPP，MQTT更加轻量级，并且占用的宽带低。
MQTT协议有以下特点：
那么问题来了？重连连接成功后重复接收到最后一条消息
MQTT推送消息订阅端重复接收问题。
（背景）订阅端断开的时候，发布端多次推送消息。
（现象）订阅端启动时，接收到最后一条推送消息有两次；即使Qos设置为2；依然是两次。
经排查是因为
MqttMessage的Retained设置为了true
该值很多文章上只说了是 消息保留机制，若设置为true，mqtt服务器会保留每次发布的消息；较少提到 若订阅某主题的客户端重启，则会把此主题之前发布的消息重新推送到客户端。该值默认为false去掉修改该值即可
那么问题来了？重连连接后手动那么多遗漏的消息，怎么选择只接收最新的一条消息呢？
MQTT推送消息订阅端重复接收问题。
（背景）订阅端断开的时候，发布端多次推送消息。
（现象）订阅端启动时，接收到msg1，msg2，msg3 （这三个消息都是同一个类型消息，只需要处理最新的msg3就好，不然界面会刷新三次）这个谁有什么好办法没呢？
GitHub地址： https://github.com/eclipse/paho.mqtt.android
mqtt的官方文档： http://mqtt.org/documentation
Github上有中文翻译： https://github.com/mcxiaoke/mqtt
在module的build.gradle文件中添加依赖
在 AndroidManifest.xml 添加限权
在 AndroidManifest.xml 注册Service （MyMqttService为自己写的服务，下文会讲到）
1、 下载Apollo服务器，下载后解压，然后运行apache-apollo-1.6\bin\apollo.cmd，输入create mybroker（名字任意取，这里是根据 官网 介绍的来取的）创建服务器实例，服务器实例包含了所有的配置，运行时数据等，并且和一个服务器进程关联。2、create mybroker之后会在bin目录下生成mybroker文件夹，里面包含有很多信息，其中etc\apollo.xml文件下是配置服务器信息的文件，etc\users.properties文件包含连接MQTT服务器时用到的用户名和密码，后面会介绍，可以修改原始的admin=password，可以接着换行添加新的用户名密码。
3、打开cmd，运行…apache-apollo-1.6\bin\mybroker\bin\apollo-broker.cmd run 开启服务器，可以在浏览器中输入 http://127.0.0.1:61680/ 查看是否安装成功，该界面展示了topic，连接数等很多信息。
经过上面的简单步骤，服务器基本上就已经完成，下一篇将介绍Android客户端的编写和注意事项。
客户端使用的API，开始我使用的是mqtt-client，使用过后发现问题百出，不能很好的满足要求，后来使用了官方推荐的 Eclipse Paho ，下面开始客户端代码的编写，为了方便测试这里有android和j2se两个工程：
1、新建android工程MQTTClient
2、MainActivity代码如下：
[java] view plaincopyprint?
package ldw.mqttclient
import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken
import org.eclipse.paho.client.mqttv3.MqttCallback
import org.eclipse.paho.client.mqttv3.MqttClient
import org.eclipse.paho.client.mqttv3.MqttConnectOptions
import org.eclipse.paho.client.mqttv3.MqttException
import org.eclipse.paho.client.mqttv3.MqttMessage
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import android.app.Activity
import android.os.Bundle
import android.os.Handler
import android.os.Message
import android.view.KeyEvent
import android.widget.TextView
import android.widget.Toast
public class MainActivity extends Activity {
private TextView resultTv
private String host = "tcp://127.0.0.1:1883"
private String userName = "admin"
private String passWord = "password"
private Handler handler
private MqttClient client
private String myTopic = "test/topic"
private MqttConnectOptions options
private ScheduledExecutorService scheduler
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState)
setContentView(R.layout.main)
resultTv = (TextView) findViewById(R.id.result)
init()
handler = new Handler() {
@Override
public void handleMessage(Message msg) {
super.handleMessage(msg)
if(msg.what == 1) {
Toast.makeText(MainActivity.this, (String) msg.obj,
Toast.LENGTH_SHORT).show()
System.out.println("-----------------------------")
} else if(msg.what == 2) {
Toast.makeText(MainActivity.this, "连接成功", Toast.LENGTH_SHORT).show()
try {
client.subscribe(myTopic, 1)
} catch (Exception e) {
e.printStackTrace()
}
} else if(msg.what == 3) {
Toast.makeText(MainActivity.this, "连接失败，系统正在重连", Toast.LENGTH_SHORT).show()
}
}
}
startReconnect()
}
private void startReconnect() {
scheduler = Executors.newSingleThreadScheduledExecutor()
scheduler.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
if(!client.isConnected()) {
connect()
}
}
}, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS)
}
private void init() {
try {
//host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
client = new MqttClient(host, "test",
new MemoryPersistence())
//MQTT的连接设置
options = new MqttConnectOptions()
//设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
options.setCleanSession(true)
//设置连接的用户名
options.setUserName(userName)
//设置连接的密码
options.setPassword(passWord.toCharArray())
// 设置超时时间 单位为秒
options.setConnectionTimeout(10)
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
options.setKeepAliveInterval(20)
//设置回调
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
//连接丢失后，一般在这里面进行重连
System.out.println("connectionLost----------")
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
//publish后会执行到这里
System.out.println("deliveryComplete---------"
+ token.isComplete())
}
@Override
public void messageArrived(String topicName, MqttMessage message)
throws Exception {
//subscribe后得到的消息会执行到这里面
System.out.println("messageArrived----------")
Message msg = new Message()
msg.what = 1
msg.obj = topicName+"---"+message.toString()
handler.sendMessage(msg)
}
})
// connect()
} catch (Exception e) {
e.printStackTrace()
}
}
private void connect() {
new Thread(new Runnable() {
@Override
public void run() {
try {
client.connect(options)
Message msg = new Message()
msg.what = 2
handler.sendMessage(msg)
} catch (Exception e) {
e.printStackTrace()
Message msg = new Message()
msg.what = 3
handler.sendMessage(msg)
}
}
}).start()
}
@Override
public boolean onKeyDown(int keyCode, KeyEvent event) {
if(client != null &&keyCode == KeyEvent.KEYCODE_BACK) {
try {
client.disconnect()
} catch (Exception e) {
e.printStackTrace()
}
}
return super.onKeyDown(keyCode, event)
}
@Override
protected void onDestroy() {
super.onDestroy()
try {
scheduler.shutdown()
client.disconnect()
} catch (MqttException e) {
e.printStackTrace()
}
}
}
由于项目需要，我用到了心跳重连。根据 这里 的解释设置apollo.xml，主要有设置主机连接的地址。另外，options还有个setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。
主要讲下Android如何使用MQTT通讯。用到的软件或者框架有：
EMQ： https://www.emqx.io/cn/
org.eclipse.paho的MQTT通讯框架： https://github.com/eclipse/paho.mqtt.android
如果已经有MQTT相关服务，可以跳过第一项，从第二项开始看。
1.安装所需要的依赖包
2.使用以下命令设置稳定存储库，以 CentOS7 为例
3.安装最新版本的 EMQ X
4.安装特定版本的 EMQ X
5.启动 EMQ X
地址：xxx.xxx.xxx:18083，地址为服务器ip或者域名，端口为18083端口
1.在Android中导入依赖
项目地址： https://github.com/eclipse/paho.mqtt.android
2.创建MQTT连接的一个Service
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）