一、MQTT协议介绍

客户机较小并且 MQTT 协议高效地使用网络带宽，在这个意义上，其为轻量级。MQTT 协议支持可靠的传送和即发即弃的传输。 在此协议中，消息传送与应用程序脱离。 脱离应用程序的程度取决于写入 MQTT 客户机和 MQTT 服务器的方式。脱离式传送能够将应用程序从任何服务器连接和等待消息中解脱出来。 交互模式与电子邮件相似，但在应用程序编程方面进行了优化。

二、MQTT优点

1.一对多消息分发外，发布/预订也脱离了应用程序。对于具有多个客户机的应用程序来说，这些功能非常有用。

2.它与消息内容没有任何关系。

3.它通过 TCP/IP 运行，TCP/IP 可以提供基本网络连接。

4.它针对消息传送提供三种服务质量：

“至多一次”

消息根据底层因特网协议网络尽最大努力进行传递。 可能会丢失消息。

例如，将此服务质量与通信环境传感器数据一起使用。 对于是否丢失个别读取或是否稍后立即发布新的读取并不重要。

“至少一次”

保证消息抵达，但可能会出现重复。

“刚好一次”

确保只收到一次消息。

例如，将此服务质量与记帐系统一起使用。

重复或丢失消息可能会导致不便或收取错误费用。

三、相关资料

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 Android客户端具体实现

基本概念：

topic：中文意思是“话题”。在MQTT中订阅了(subscribe)同一话题(topic)的客户端会同时收到消息推送。直接实现了“群聊”功能。

clientId：客户身份唯一标识。

qos：服务质量。

retained：要保留最后的断开连接信息。

MqttAndroidClient#subscribe()：订阅某个话题。

MqttAndroidClient#publish()： 向某个话题发送消息，之后服务器会推送给所有订阅了此话题的客户。

userName：连接到MQTT服务器的用户名。

passWord ：连接到MQTT服务器的密码。

添加依赖

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'

}

添加限权

注册

Service

创建service

public class MQTTService extends Service {

public static final String TAG = MQTTService.class.getSimpleName();

private static MqttAndroidClient client;

private MqttConnectOptions conOpt;

private String host = "tcp://192.168.229.80:1883";

private String userName = "admin";

private String passWord = "password";

private static String myTopic = "gmt/report/" + UserConfig.get().getAccount();//要订阅的主题

private static String sendTopic = "gmt/control/" + UserConfig.get().getmGateway();//要发布控制主题

private String clientId = "152xxxx2901";//客户端标识

private IGetMessageCallBack mGetMessageCallBack;

static IGetMessageCallBack mcallBack;

@Override

public void onCreate() {

super.onCreate();

Log.e(getClass().getName(), "onCreate");

init();

}

public static void publish(String msg) {

String topic = sendTopic;

Integer qos = 0;

Boolean retained = false;

try {

if (client != null) {

client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());

}

} catch (MqttException e) {

e.printStackTrace();

}

}

public static void publish(String msg, IGetMessageCallBack callBack) {

mcallBack = callBack;

String topic = sendTopic;

Integer qos = 0;

Boolean retained = false;

if (client != null) {

try {

client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());

} catch (MqttException e) {

e.printStackTrace();

}

}

}

private void init() {

// 服务器地址(协议+地址+端口号)

String uri = host;

client = new MqttAndroidClient(this, uri, clientId);

// 设置MQTT监听并且接受消息

client.setCallback(mqttCallback);

conOpt = new MqttConnectOptions();

// 清除缓存

conOpt.setCleanSession(true);

// 设置超时时间，单位：秒

conOpt.setConnectionTimeout(10);

// 心跳包发送间隔，单位：秒

conOpt.setKeepAliveInterval(20);

// 用户名

conOpt.setUserName(userName);

// 密码

conOpt.setPassword(passWord.toCharArray()); //将字符串转换为字符串数组

// last will message

boolean doConnect = true;

String message = "{\"terminal_uid\":\"" + clientId + "\"}";

Log.e(getClass().getName(), "message是:" + message);

String topic = myTopic;

Integer qos = 0;

Boolean retained = false;

if ((!message.equals("")) || (!topic.equals(""))) {

// 最后的遗嘱

// MQTT本身就是为信号不稳定的网络设计的，所以难免一些客户端会无故的和Broker断开连接。

//当客户端连接到Broker时，可以指定LWT，Broker会定期检测客户端是否有异常。

//当客户端异常掉线时，Broker就往连接时指定的topic里推送当时指定的LWT消息。

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();

}

}

@Override

public void onDestroy() {

// 此处解决MQTT退出异常问题

if (client != null) {

client.unregisterResources();

client.close();

}

super.onDestroy();

}

/**

* 连接MQTT服务器

*/

private void doClientConnection() {

if (!client.isConnected() && isConnectIsNormal()) {

try {

client.connect(conOpt, null, iMqttActionListener);

} catch (MqttException e) {

e.printStackTrace();

}

}

}

// MQTT是否连接成功

private IMqttActionListener iMqttActionListener = new IMqttActionListener() {

@Override

public void onSuccess(IMqttToken arg0) {

Log.i(TAG, "连接成功 ");

try {

// 订阅myTopic话题

Log.i(TAG, "SUB:" + myTopic);

client.subscribe(myTopic, 0);

} catch (MqttException e) {

e.printStackTrace();

Log.e(TAG, "sube：" + e.getMessage());

}

}

@Override

public void onFailure(IMqttToken arg0, Throwable arg1) {

arg1.printStackTrace();

// 连接失败，重连

}

};

// MQTT监听并且接受消息

private MqttCallback mqttCallback = new MqttCallback() {

@Override

public void messageArrived(String topic, MqttMessage message) throws Exception {

String str1 = new String(message.getPayload());

JSONObject jsonObject = new JSONObject(str1);

JSONObject msg = jsonObject.optJSONObject("msg");

if (mGetMessageCallBack != null && msg != null) {

mGetMessageCallBack.setMessage(msg);

mcallBack.setMessage(msg);

}

String str2 = topic + ";qos:" + message.getQos() + ";retained:" + message.isRetained();

Log.i(TAG, "messageArrived:" + str1);

Log.i(TAG, str2);

}

@Override

public void deliveryComplete(IMqttDeliveryToken arg0) {

}

@Override

public void connectionLost(Throwable arg0) {

// 失去连接，重连

}

};

/**

* 判断网络是否连接

*/

private boolean isConnectIsNormal() {

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);

return true;

} else {

Log.i(TAG, "MQTT 没有可用网络");

return false;

}

}

@Override

public IBinder onBind(Intent intent) {

Log.e(getClass().getName(), "onBind");

return new CustomBinder();

}

public void setIGetMessageCallBack(IGetMessageCallBack IGetMessageCallBack) {

this.mGetMessageCallBack = IGetMessageCallBack;

}

public class CustomBinder extends Binder {

public MQTTService getService() {

return MQTTService.this;

}

}

@Override

public boolean onUnbind(Intent intent) {

return super.onUnbind(intent);

}

}

接口回调

public interface IGetMessageCallBack {

public void setMessage(JSONObject message);

}

代码使用

public class MainActivity extends BaseActivity implements IGetMessageCallBack {

@Override

protected int initLayout() {

return R.layout.activity_main;

}

@Override

protected void initView() {

initBind();

// 此处注册方便回调

getServiceConnection().setIGetMessageCallBack(this);

}

@Override

protected void onDestroy() {

super.onDestroy();

}

// 此处接收消息

@Override

public void setMessage(JSONObject message) {

Log.e("bingo", "main:" + message);

}

}

注意要在BaseActivity中封装如下

public abstract class BaseActivity extends AppCompatActivity {

private Context mContext;

private Handler handler;

private MyServiceConnection serviceConnection;

public MyServiceConnection getServiceConnection() {

return serviceConnection;

}

@Override

public void onCreate(@Nullable Bundle savedInstanceState) {

super.onCreate(savedInstanceState);

setContentView(initLayout());

mContext = this;

initConnectTion();

}

// 此处申请绑定

private void initConnectTion() {

serviceConnection = new MyServiceConnection();

Intent intent = new Intent(this, MQTTService.class);

bindService(intent, serviceConnection, Context.BIND_AUTO_CREATE);

}

@Override

protected void onResume() {

super.onResume();

}

public Context getContext() {

return mContext;

}

@Override

protected void onDestroy() {

// 要在此处进行解绑

if (serviceConnection != null) {

unbindService(serviceConnection);

}

super.onDestroy();

}

}

异常