本文已授权微信公众号《鸿洋》原创首发，转载请务必注明出处。
MQTT官网： http://mqtt.org/
MQTT介绍： http://www.ibm.com
MQTT Android github： https://github.com/eclipse/paho.mqtt.android
MQTT API： http://www.eclipse.org/paho/files/javadoc/index.html
MQTT Android API： http://www.eclipse.org/paho/files/android-javadoc/index.html
建议时间充裕的同学有顺序的阅读上文五个链接内容，不充裕的同学请看下面简单的介绍（内容大多来自上面五条链接）。
MQTT 协议
客户机较小并且 MQTT 协议 高效地使用网络带宽，在这个意义上，其为轻量级。MQTT 协议支持可靠的传送和即发即弃的传输。 在此协议中，消息传送与应用程序脱离。 脱离应用程序的程度取决于写入 MQTT 客户机和 MQTT 服务器的方式。脱离式传送能够将应用程序从任何服务器连接和等待消息中解脱出来。 交互模式与电子邮件相似，但在应用程序编程方面进行了优化。
协议具有许多不同的功能：
之后在浏览器输入 http://127.0.0.1:61680/，查看是否安装成功。
subscribe)同一话题（ topic）的客户端会同时收到消息推送。直接实现了“群聊”功能。 repositories {
maven {
url"https://repo.eclipse.org/content/repositories/paho-releases/"}
}
dependencies {
compile'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'compile'org.eclipse.paho:org.eclipse.paho.android.service:1.1.0'} <uses-permissionandroid:name="android.permission.INTERNET"/><uses-permissionandroid:name="android.permission.ACCESS_NETWORK_STATE"/><uses-permissionandroid:name="android.permission.WAKE_LOCK"/> <!-- Mqtt Service --><serviceandroid:name="org.eclipse.paho.android.service.MqttService"/><serviceandroid:name="com.dongyk.service.MQTTService"/> packagecom.dongyk.service;importandroid.app.Service;
.../**
* MQTT长连接服务
*
*@author一口仨馍 联系方式 : [email protected]
*@version创建时间：2016/9/16 22:06
*/publicclassMQTTServiceextendsService{publicstaticfinalString TAG = MQTTService.class.getSimpleName();privatestaticMqttAndroidClient client;privateMqttConnectOptions conOpt;// private String host = "tcp://10.0.2.2:61613";privateString host ="tcp://192.168.1.103:61613";privateString userName ="admin";privateString passWord ="password";privatestaticString myTopic ="topic";privateString clientId ="test";@OverridepublicintonStartCommand(Intent intent,intflags,intstartId) {
init();returnsuper.onStartCommand(intent, flags, startId);
}publicstaticvoidpublish(String msg){
String topic = myTopic;
Integer qos =0;
Boolean retained =false;try{
client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());
}catch(MqttException e) {
e.printStackTrace();
}
}privatevoidinit() {// 服务器地址（协议+地址+端口号）String uri = host;
client =newMqttAndroidClient(this, uri, clientId);// 设置MQTT监听并且接受消息client.setCallback(mqttCallback);
conOpt =newMqttConnectOptions();// 清除缓存conOpt.setCleanSession(true);// 设置超时时间，单位：秒conOpt.setConnectionTimeout(10);// 心跳包发送间隔，单位：秒conOpt.setKeepAliveInterval(20);// 用户名conOpt.setUserName(userName);// 密码conOpt.setPassword(passWord.toCharArray());// last will messagebooleandoConnect =true;
String message ="{\"terminal_uid\":\""+ clientId +"\"}";
String topic = myTopic;
Integer qos =0;
Boolean retained =false;if((!message.equals("")) || (!topic.equals(""))) {// 最后的遗嘱try{
conOpt.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
}catch(Exception e) {
Log.i(TAG,"Exception Occured", e);
doConnect =false;
iMqttActionListener.onFailure(null, e);
}
}if(doConnect) {
doClientConnection();
}
}@OverridepublicvoidonDestroy() {try{
client.disconnect();
}catch(MqttException e) {
e.printStackTrace();
}super.onDestroy();
}/** 连接MQTT服务器 */privatevoiddoClientConnection() {if(!client.isConnected() && isConnectIsNomarl()) {try{
client.connect(conOpt,null, iMqttActionListener);
}catch(MqttException e) {
e.printStackTrace();
}
}
}// MQTT是否连接成功privateIMqttActionListener iMqttActionListener =newIMqttActionListener() {@OverridepublicvoidonSuccess(IMqttToken arg0) {
Log.i(TAG,"连接成功 ");try{// 订阅myTopic话题client.subscribe(myTopic,1);
}catch(MqttException e) {
e.printStackTrace();
}
}@OverridepublicvoidonFailure(IMqttToken arg0, Throwable arg1) {
arg1.printStackTrace();// 连接失败，重连}
};// MQTT监听并且接受消息privateMqttCallback mqttCallback =newMqttCallback() {@OverridepublicvoidmessageArrived(String topic, MqttMessage message)throwsException {
String str1 =newString(message.getPayload());
MQTTMessage msg =newMQTTMessage();
msg.setMessage(str1);
EventBus.getDefault().post(msg);
String str2 = topic +";qos:"+ message.getQos() +";retained:"+ message.isRetained();
Log.i(TAG,"messageArrived:"+ str1);
Log.i(TAG, str2);
}@OverridepublicvoiddeliveryComplete(IMqttDeliveryToken arg0) {
}@OverridepublicvoidconnectionLost(Throwable arg0) {// 失去连接，重连}
};/** 判断网络是否连接 */privatebooleanisConnectIsNomarl() {
ConnectivityManager connectivityManager = (ConnectivityManager)this.getApplicationContext().getSystemService(Context.CONNECTIVITY_SERVICE);
NetworkInfo info = connectivityManager.getActiveNetworkInfo();if(info !=null&& info.isAvailable()) {
String name = info.getTypeName();
Log.i(TAG,"MQTT当前网络名称："+ name);returntrue;
}else{
Log.i(TAG,"MQTT 没有可用网络");returnfalse;
}
}@Nullable@OverridepublicIBinderonBind(Intent intent) {returnnull;
}
} 首先初始化各个参数，之后连接服务器。连接成功之后在 http://127.0.0.1:61680/看到自动创建了名称为”topic”的 topic。这里我使用了一个真机和一个模拟器运行程序。 http://127.0.0.1:61680/服务端看到的是这个样子
这里需要注意两个地方
1. 模拟器运行的时候 host = "tcp://10.0.2.2:61613"，因为10.0.2.2 是模拟器设置的特定ip，是你电脑的别名。真机运行的时候 host = "tcp://192.168.1.103:61613"。192.168.1.103是我主机的IPv4地址，查看本机IP的cmd命令为 ipconfig/all。
2. 两次运行时的 clientId不能一样（为了保证客户标识的唯一性）。
这里为了测试，在 MQTTService中暴露了一个公共方法 publish(String msg)给 MainActivity调用。代码如下
publicclassMainActivityextendsActivity{@OverrideprotectedvoidonCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);
setContentView(R.layout.activity_main);
EventBus.getDefault().register(this);
startService(newIntent(this, MQTTService.class));
findViewById(R.id.publishBtn).setOnClickListener(newView.OnClickListener() {@OverridepublicvoidonClick(View view) {
MQTTService.publish("CSDN 一口仨馍");
}
});
}@Subscribe(threadMode = ThreadMode.MAIN)publicvoidgetMqttMessage(MQTTMessage mqttMessage){
Log.i(MQTTService.TAG,"get message:"+mqttMessage.getMessage());
Toast.makeText(this,mqttMessage.getMessage(),Toast.LENGTH_SHORT).show();
}@OverrideprotectedvoidonDestroy() {
EventBus.getDefault().unregister(this);super.onDestroy();
}
} 这里使用了 EventBus3.0发送消息，感兴趣的可以看下 EventBus3.0使用及源码解析。当然，你也可以使用接口回调的方式甚至直接在Service中弹出 Toast。whatever，现在点击一个客户端 MainActivity中的 Button，两个客户端已经能同时弹出消息。已经 get到数据了。接下来，show time~