笔者在上一章对连接报文进行了相关的讲解。这一章笔者想写一个连接报文的例子来加深理解。本来这一章也应该在上一章出现的。可是笔者怕太长了。不好方便阅。所以决定分俩章来。正如笔者上一章所讲的。笔者会用Netty通信框架进行编写。主要因为Netty已经为我们集成了相关MQTT功能。
开发环境
开发工具：intellij idea.(以前我一直在eclipse。最近新版的老报错。所以就放弃了)
Netty包：netty-all-4.1.16.Final.jar。下载网站：http://netty.io/downloads.html
JDK：JAVA 8
第三包：commons-lang3-3.6.jar。下载网站：http://commons.apache.org/proper/commons-lang/download_lang.cgi
MQTT编写
在这里笔者并不打包把客户端的代码一起编写出。事实上关于客户端的开源的代码是非常多的。笔者这里只会略微的编写一下服务端的代码。当然这里代码只是为方更了解MQTT协议。并非企业级的编蜜枣这一点希望读者见谅。为了实现连接报文。笔者定义了三个类。
Main类：用于启动服务。
BrokerHandler类：处理接受来的信息。
BrokerSessionHelper：用于发送信息给客户。
Main类的源码
1 public static void main(String[] args) throws Exception { 2 EventLoopGroup bossGroup = new NioEventLoopGroup(); 3 EventLoopGroup workerGroup = new NioEventLoopGroup(); 4 5 Runtime.getRuntime().addShutdownHook(new Thread() { 6 public void run() { 7 workerGroup.shutdownGracefully(); 8 bossGroup.shutdownGracefully(); 9 } 10 }); 11 12 13 ServerBootstrap b = new ServerBootstrap(); 14 b.group(bossGroup, workerGroup) 15 .channel(NioServerSocketChannel.class) 16 .handler(new LoggingHandler(LogLevel.INFO)) 17 .childHandler(new ChannelInitializer<SocketChannel>() { 18 @Override 19 public void initChannel(SocketChannel ch) throws Exception { 20 21 ChannelPipeline p = ch.pipeline(); 22 23 p.addFirst("idleHandler", new IdleStateHandler(0, 0, 120)); 24 p.addLast("encoder", MqttEncoder.INSTANCE); 25 p.addLast("decoder", new MqttDecoder()); 26 p.addLast("logicHandler", new BrokerHandler(65535)); 27 28 } 29 }) 30 .option(ChannelOption.SO_BACKLOG, 511) 31 .childOption(ChannelOption.SO_KEEPALIVE, true); 32 33 ChannelFuture f = b.bind("0.0.0.0", 1883).sync(); 34 35 f.channel().closeFuture().sync(); 36 }
上面的1、2俩行表是Netty里面俩个线程组。事实上也就是Reactor线程组。bossGroup 用于处理接受来自客户端的连接。workerGroup 用于处理接受客户端的读取信息。13行的ServerBootstrap可以理解为启动服务的一个引导类。主要关键是他的group方法。这样子就可以把俩个线程组关系在一起了。重点就在17行这里。childHandler用于处理IO事件。比如读取客户端进行。然后自己编码。你们可以看到24行的MqttEncoder.INSTANCE和25行的MqttDecoder吧。他们就是用于处理MQTT协议传来的信息进行处理。而26行BrokerHandler类就是笔者来处理每一个报文对应的响应。笔者就不在过多的说了。如果你们不懂的话，可以去看一下Netty框架的知识在过看的话会比较好。
BrokerHandler类的源码
1 public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> { 2 private MqttVersion version; 3 private String clientId; 4 private String userName; 5 private String brokerId; 6 private boolean connected; 7 private boolean cleanSession; 8 private int keepAlive; 9 private int keepAliveMax; 10 private MqttPublishMessage willMessage; 11 12 public BrokerHandler(int keepAliveMax) { 13 14 this.keepAliveMax = keepAliveMax; 15 } 16 17 @Override 18 @SuppressWarnings("ThrowableResultOfMethodCallIgnored") 19 protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception { 20 21 if (msg.decoderResult().isFailure()) { 22 23 Throwable cause = msg.decoderResult().cause(); 24 25 if (cause instanceof MqttUnacceptableProtocolVersionException) { 26 27 BrokerSessionHelper.sendMessage( 28 ctx, 29 MqttMessageFactory.newMessage( 30 new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0), 31 new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false), 32 null), 33 "INVALID", 34 null, 35 true); 36 37 } else if (cause instanceof MqttIdentifierRejectedException) { 38 39 BrokerSessionHelper.sendMessage( 40 ctx, 41 MqttMessageFactory.newMessage( 42 new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0), 43 new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), 44 null), 45 "INVALID", 46 null, 47 true); 48 } 49 50 ctx.close(); 51 52 return; 53 } 54 55 switch (msg.fixedHeader().messageType()) { 56 case CONNECT: 57 onConnect(ctx, (MqttConnectMessage) msg); 58 break; 59 case PUBLISH: 60 onPublish(ctx, (MqttPublishMessage) msg); 61 break; 62 case PUBACK: 63 onPubAck(ctx, msg); 64 break; 65 case PUBREC: 66 onPubRec(ctx, msg); 67 break; 68 case PUBREL: 69 onPubRel(ctx, msg); 70 break; 71 case PUBCOMP: 72 onPubComp(ctx, msg); 73 break; 74 case SUBSCRIBE: 75 onSubscribe(ctx, (MqttSubscribeMessage) msg); 76 break; 77 case UNSUBSCRIBE: 78 onUnsubscribe(ctx, (MqttUnsubscribeMessage) msg); 79 break; 80 case PINGREQ: 81 onPingReq(ctx); 82 break; 83 case DISCONNECT: 84 onDisconnect(ctx); 85 break; 86 } 87 88 } 89 90 private void onConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) { 91 92 this.version = MqttVersion.fromProtocolNameAndLevel(msg.variableHeader().name(), (byte) msg.variableHeader().version()); 93 this.clientId = msg.payload().clientIdentifier(); 94 this.cleanSession = msg.variableHeader().isCleanSession(); 95 96 if (msg.variableHeader().keepAliveTimeSeconds() > 0 && msg.variableHeader().keepAliveTimeSeconds() <= this.keepAliveMax) { 97 this.keepAlive = msg.variableHeader().keepAliveTimeSeconds(); 98 } 99 100 //MQTT 3.1之后可能存在为空的客户ID。所以要进行处理。如果客户ID是空，而且还在保存处理相关的信息。这样子是不行。 101 //必须有客户ID我们才能存保相关信息。 102 if (StringUtils.isBlank(this.clientId)) { 103 if (!this.cleanSession) { 104 105 BrokerSessionHelper.sendMessage( 106 ctx, 107 MqttMessageFactory.newMessage( 108 new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0), 109 new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), 110 null), 111 "INVALID", 112 null, 113 true); 114 115 ctx.close(); 116 117 return; 118 119 } else { 120 this.clientId = java.util.UUID.randomUUID().toString(); 121 } 122 } 123 124 //有可能发送俩次的连接包。如果已经存在连接就是关闭当前的连接。 125 if (this.connected) { 126 ctx.close(); 127 return; 128 } 129 130 131 boolean userNameFlag = msg.variableHeader().hasUserName(); 132 boolean passwordFlag = msg.variableHeader().hasPassword(); 133 this.userName = msg.payload().userName(); 134 135 String password = "" ; 136 if( msg.payload().passwordInBytes() != null && msg.payload().passwordInBytes().length > 0) 137 password = new String(msg.payload().passwordInBytes()); 138 139 boolean mistake = false; 140 141 //如果有用户名标示，那么就必须有密码标示。 142 //当有用户名标的时候，用户不能为空。 143 //当有密码标示的时候，密码不能为空。 144 if (userNameFlag) { 145 if (StringUtils.isBlank(this.userName)) 146 mistake = true; 147 } else { 148 if (StringUtils.isNotBlank(this.userName) || passwordFlag) mistake = true; 149 } 150 151 152 if (passwordFlag) { 153 154 if (StringUtils.isBlank(password)) mistake = true; 155 } else { 156 if (StringUtils.isNotBlank(password)) mistake = true; 157 } 158 159 if (mistake) { 160 BrokerSessionHelper.sendMessage( 161 ctx, 162 MqttMessageFactory.newMessage( 163 new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0), 164 new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false), 165 null), 166 this.clientId, 167 null, 168 true); 169 ctx.close(); 170 return; 171 } 172 173 BrokerSessionHelper.sendMessage( 174 ctx, 175 MqttMessageFactory.newMessage( 176 new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0), 177 new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, !this.cleanSession), 178 null), 179 this.clientId, 180 null, 181 true); 182 183 ChannelHandlerContext lastSession = BrokerSessionHelper.removeSession(this.clientId); 184 if (lastSession != null) { 185 lastSession.close(); 186 } 187 188 String willTopic = msg.payload().willTopic(); 189 String willMessage = ""; 190 if(msg.payload().willMessageInBytes() != null && msg.payload().willMessageInBytes().length > 0) 191 willMessage = new String(msg.payload().willMessageInBytes()); 192 193 if (msg.variableHeader().isWillFlag() && StringUtils.isNotEmpty(willTopic) && StringUtils.isNotEmpty(willMessage)) { 194 195 this.willMessage = (MqttPublishMessage) MqttMessageFactory.newMessage( 196 new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(msg.variableHeader().willQos()), msg.variableHeader().isWillRetain(), 0), 197 new MqttPublishVariableHeader(willTopic, 0), 198 Unpooled.wrappedBuffer(willMessage.getBytes()) 199 ); 200 } 201 202 this.connected = true; 203 BrokerSessionHelper.saveSession(this.clientId, ctx); 204 } 205 206 private void onSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) { 207 } 208 209 private void onUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) { 210 } 211 212 private void onPingReq(ChannelHandlerContext ctx) { 213 } 214 215 private void onDisconnect(ChannelHandlerContext ctx) { 216 217 if (!this.connected) { 218 ctx.close(); 219 return; 220 } 221 222 BrokerSessionHelper.removeSession(this.clientId, ctx); 223 224 this.willMessage = null; 225 226 this.connected = false; 227 228 ctx.close(); 229 230 } 231 232 private void onPubComp(ChannelHandlerContext ctx, MqttMessage msg) { 233 234 } 235 236 private void onPubRel(ChannelHandlerContext ctx, MqttMessage msg) { 237 } 238 239 private void onPubRec(ChannelHandlerContext ctx, MqttMessage msg) { 240 } 241 242 private void onPubAck(ChannelHandlerContext ctx, MqttMessage msg) { 243 } 244 245 private void onPublish(ChannelHandlerContext ctx, MqttPublishMessage msg) { 246 } 247 }