前言
MQTT定义了物联网传输协议，其标准倾向于原始TCP实现。构建于TCP的上层协议堆栈，诸如HTTP等，在空间上多了一些处理路径，稍微耗费了CPU和内存，虽看似微乎其微，但对很多处理能力不足的嵌入式设备而言，选择原始的TCP却是最好的选择。
但单纯TCP不是所有物件联网的最佳选择，提供构建与TCP基础之上的传统的HTTP通信支持，尤其是浏览器、性能富裕的桌面涉及领域，还是企业最 可信赖、最可控的传输方式之一。支持多种多样的连接通道，让目前所有一切皆可联网，除了原始TCP Socket，还要支持构建于其之上的HTTP、HTML5 Websocket，就很有必要。
mqtt.io，Pub/Sub中间件，也可以称之为推送服务器，涵盖所有主流桌面系统、浏览器平台，并且倾斜 于移动互联网，以及物联网的广阔适应天地。使用一句英文概括可能更为合适："Make everything connect”，让所有物件都可连接。其业务目标，可用下图概括：
mqtt.io致力于做下一代支持所有主流桌面平台、所有主流浏览器、所有可联网物件都可以联网的PUB/SUB消息推送系统。
构建此系统，在于降低传统企业各自分散的推送系统，统一运营，统一管理，节省人员、运维开支。
注意事项
- mqtt.io是一个项目名称，没有官网，http://www.mqtt.io，和这个项目没有一毛钱关系。
- 项目地址：https://github.com/yongboy/mqtt.io，
- 项目名称启发于 http://socket.io http://netty.io 等知名framework。
- 目前只实现QoS 0基本特性，实现概览，后期会根据反馈，做出一些调整
依赖
- netty 4，目前JAVA IO界明星
- mqtt-library 二进制和MQTT对象的转换，这种苦活累活都是它来做，真心让人喜欢。
数据流转
解码器
用于转换二进制流到JAVA对象的过程：
| 51 |
io.mqtt.handler.coder;
io.netty.buffer.ByteBuf;
io.netty.channel.ChannelHandlerContext;
io.netty.handler.codec.MessageToMessageDecoder;
java.io.ByteArrayInputStream;
java.util.List;
org.meqantt.message.Message;
org.meqantt.message.MessageInputStream;
ByteBuf> {
@Override
buf,
Exception {
2) {
return;
}
.markReaderIndex();
// read away header
0;
1;
int digit;
0;
do {
++;
.readByte();
* multiplier;
128;
.isReadable()) {
.resetReaderIndex();
return;
}
0);
< msgLength) {
.resetReaderIndex();
return;
}
+ msgLength];
.resetReaderIndex();
.readBytes(data);
MessageInputStream(
ByteArrayInputStream(data));
.readMessage();
.close();
.add(msg);
}
}
|
编码器
对所有要写入网卡缓冲区的JAVA对象转换成二进制：
| 25 |
io.mqtt.handler.coder;
io.netty.buffer.Unpooled;
io.netty.channel.ChannelHandler.Sharable;
io.netty.channel.ChannelHandlerContext;
io.netty.handler.codec.MessageToMessageEncoder;
java.util.List;
org.meqantt.message.Message;
Sharable
Object> {
@Override
msg,
Exception {
Message)) {
return;
}
.toBytes();
.wrappedBuffer(data));
}
}
|
借助于mqtt-library项目，编解码不复杂。
MQTT的消息处理
| 89 |
io.mqtt.handler;
io.mqtt.processer.ConnectProcesser;
io.mqtt.processer.DisConnectProcesser;
io.mqtt.processer.PingReqProcesser;
io.mqtt.processer.Processer;
io.mqtt.processer.PublishProcesser;
io.mqtt.processer.SubscribeProcesser;
io.mqtt.processer.UnsubscribeProcesser;
io.netty.channel.ChannelFutureListener;
io.netty.channel.ChannelHandlerContext;
io.netty.channel.ChannelInboundHandlerAdapter;
io.netty.handler.timeout.ReadTimeoutException;
java.util.Collections;
java.util.HashMap;
java.util.Map;
org.meqantt.message.ConnAckMessage;
org.meqantt.message.ConnAckMessage.ConnectionStatus;
org.meqantt.message.DisconnectMessage;
org.meqantt.message.Message;
org.meqantt.message.Message.Type;
org.meqantt.message.PingRespMessage;
ChannelInboundHandlerAdapter {
PingRespMessage();
> processers;
static {
Processer>(
6);
ConnectProcesser());
PublishProcesser());
SubscribeProcesser());
UnsubscribeProcesser());
PingReqProcesser());
DisConnectProcesser());
.unmodifiableMap(map);
}
@Override
e)
Exception {
try {
ReadTimeoutException) {
.addListener(
.CLOSE_ON_FAILURE);
else {
.close();
}
Throwable t) {
.printStackTrace();
.close();
}
.printStackTrace();
}
@Override
obj)
Exception {
Message) obj;
.getType());
null) {
return;
}
.proc(msg, ctx);
null) {
return;
}
ConnAckMessage
.ACCEPTED) {
.CLOSE);
DisconnectMessage) {
.CLOSE);
else {
.CLOSE_ON_FAILURE);
}
}
@Override
Exception {
.flush();
}
}
|
更具体的可以查看项目。
小结
简单介绍了一个简单的不能再简单的MQTT Server，只具有最基本的QoS 0类型的消息订阅等。
后面，对HTML 5 Websocket，会在现有基础代码之上，不做多大改动，增加对MQTT Over WebSocket的支持。