MQTT通讯协议简介

WebSphere MQ 遥测传输 (MQTT) 是轻量级基于代理的发布 / 订阅的消息传输协议，设计思想是开放、简单、轻量、易于实现。这些特点使它适用于受限环境。例如，但不仅限于此：

网络代价昂贵，带宽低、不可靠。
在嵌入设备中运行，处理器和内存资源有限。
该协议的特点有：

使用发布 / 订阅消息模式，提供一对多的消息发布，解除应用程序耦合。
对负载内容屏蔽的消息传输。
使用 TCP/IP 提供网络连接。
有三种消息发布服务质量：
“至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。
“至少一次”，确保消息到达，但消息重复可能会发生。
“只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。
小型传输，开销很小（固定长度的头部是 2 字节），协议交换最小化，以降低网络流量。
使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制。
IBM 通过对现有移动推送平台比较之后，对其中存在的问题和缺陷做了很好的分析。这些问题集中体现在如下方面：

在网络方面如何适应现有网络的不可靠，很好的保障数据发送可靠性
如何降低网络流量，从而节省网络成本

在移动设备方面如何降低对设备能力的要求，特别是适应计算和存储弱的设备
如何降低对设备电量的消耗，满足设备电源能力的不足

如何降低平台依赖性，真正实现跨移动设备平台

在数据方面缺少对数据安全性的保障，特别是对服务器的掌控
缺少对大量数据的监测，优化

IBM 针对上面问题，结合 MQTT 和 WebSocket，提出了更智慧的移动推送服务解决方案。



图 1. IBM 移动推送服务解决方案


mosca是一个轻量级的mqtt broker(NodeJS中间件),  安装和使用都很简单：

npm install mqtt -gd

npm install mosca -gd

mosca -v &

mosca启动后， 就可以使用mqtt client发布和订阅消息了：

NodeJS client:

发布消息：

var mqttClient = require(‘mqtt’);
var client = mqttClient.createClient(1883, ’127.0.0.26′); //host与端口
client.publish(‘test/public/topic0′, ‘public topic 0 start’);
client.publish(‘test/private/topic1′, ‘private topic 1 start’);
client.publish(‘test/private/topic2′, ‘private topic 2 start’);

订阅消息：

var mqttClient = require(‘mqtt’);
var client = mqttClient.createClient(1883, ’127.0.0.26′);
//client.subscribe(‘fuck/*’);
client.subscribe(‘test/*’);
client.on(‘message’, function(topic, message){
console.log(topic, message);
});

Java Mqtt client的例子：

package com.snda.apps.odin.talk.client;

import com.ibm.mqtt.MqttClient;
import com.ibm.mqtt.MqttException;
import com.ibm.mqtt.MqttSimpleCallback;

public class MqttClienter {
private final static String CONNECTION_STRING = “tcp://127.0.0.26:1883″;
private final static boolean CLEAN_START = true;
private final static short KEEP_ALIVE = 30;// 低耗网络，但是又需要及时获取数据，心跳30s
private final static String CLIENT_ID = “client1″;
private final static String PUBLISH_TOPICS = “test/public/topic0″;
private final static String[] TOPICS = {“test/private/topic1″, “test/private/topic2″};

private final static int[] QOS_VALUES = { 0, 2 };

private MqttClient mqttClient = null;

public MqttClienter(String i) {
try {
mqttClient = new MqttClient(CONNECTION_STRING);
SimpleCallbackHandler simpleCallbackHandler = new SimpleCallbackHandler();
mqttClient.registerSimpleHandler(simpleCallbackHandler);// 注册接收消息方法
mqttClient.connect(CLIENT_ID + i, CLEAN_START, KEEP_ALIVE);
mqttClient.subscribe(TOPICS, QOS_VALUES);// 订阅接主题
/**
* 完成订阅后，可以增加心跳，保持网络通畅，也可以发布自己的消息
*/
mqttClient.publish(PUBLISH_TOPICS, “Java客户端测试消息Hello World!”.getBytes(), QOS_VALUES[0], true);
} catch (MqttException e) {
e.printStackTrace();
}
}

// 简单回调函数，处理client接收到的主题消息
class SimpleCallbackHandler implements MqttSimpleCallback {
/**
* 当客户机和broker意外断开时触发 可以再此处理重新订阅
*/
public void connectionLost() throws Exception {
System.out.println(“客户机和broker已经断开”);
}

/**
* 客户端订阅消息后，该方法负责回调接收处理消息
*/
public void publishArrived(String topicName, byte[] payload, int Qos,
boolean retained) throws Exception {
System.out.println(“订阅主题: ” + topicName);
System.out.println(“消息数据: ” + new String(payload));
System.out.println(“消息级别(0,1,2): ” + Qos);
System.out.println(“是否是实时发送的消息(false=实时，true=服务器上保留的最后消息): ” + retained);
}
}

public static void main(String[] args) {
   new MqttClienter(“” + 1);
 }
}

此协议可以用于移动APP的消息发布，即时通讯等应用