MQTT与webSocket

Mqtt底层使用webSocket实现，通过发送http或https请求与服务端开始进行handshake，握手完成后协议将从http(https)升级成webSocket并建立长连，之后通过长连进行通信。sdk中关于websocket部分结构及handshake信息如下：

MQTT心跳包

分析心跳包首先看用于发送心跳信息的MqttPingReq.class和用于接收心跳答复的MqttPingResp.class

然后，我们需要从其发送的内容当中逆向推出其心跳包的内容。我 们先看其发送的的模块：找到public class CommsSender implements Runnable 类，看到其有一个private MqttOutputStream out;私有字段，一看这个方法，我们就能判断，这个字段就是输出流，打开public class MqttOutputStream extends OutputStream这个类，你会看到这样一个方法：

/**    * Writes an MqttWireMessage to the stream.    */   public void write(MqttWireMessage message) throws IOException, MqttException {      final String methodName = "write";      byte[] bytes = message.getHeader();      byte[] pl = message.getPayload();//    out.write(message.getHeader());//    out.write(message.getPayload());      out.write(bytes,0,bytes.length);      clientState.notifySentBytes(bytes.length);              int offset = 0;        int chunckSize = 1024;        while (offset < pl.length) {           int length = Math.min(chunckSize, pl.length - offset);           out.write(pl, offset, length);           offset += chunckSize;           clientState.notifySentBytes(length);        }                 // @TRACE 500= sent {0}       log.fine(CLASS_NAME, methodName, "500", new Object[]{message});   }

原来，其发送的是header和payload，然后我们再看看心跳包的header和payload。

MqttPingReq和MqttPingResp中都有这么个方法：

protected byte[] getVariableHeader() throws MqttException {   return new byte[0];}

往上查看其共同父类MqttWireMessage.class中的getHeader()：

public byte[] getHeader() throws MqttException {    try {        int first = ((getType() & 0x0f) << 4) ^ (getMessageInfo() & 0x0f);        byte[] varHeader = getVariableHeader();        int remLen = varHeader.length + getPayload().length;//长度为0        ByteArrayOutputStream baos = new ByteArrayOutputStream();        DataOutputStream dos = new DataOutputStream(baos);        dos.writeByte(first);//写入一个字节        dos.write(encodeMBI(remLen));//查看encodeMBI()方法实现可知此处也是只有一个字节        dos.write(varHeader);//new byte[0]        dos.flush();        return baos.toByteArray();    } catch(IOException ioe) {        throw new MqttException(ioe);    }}

protected static byte[] encodeMBI( long number) {   int numBytes = 0;   long no = number;   ByteArrayOutputStream bos = new ByteArrayOutputStream();   // Encode the remaining length fields in the four bytes   do {      byte digit = (byte)(no % 128);      no = no / 128;      if (no > 0) {         digit |= 0x80;      }      bos.write(digit);      numBytes++;   } while ( (no > 0) && (numBytes<4) );      return bos.toByteArray();}

而MqttWireMessage中还有一个getPayload方法，MqttPingReq和MqttPingResp都没有重写这个方法：

/** * Sub-classes should override this method to supply the payload bytes. */public byte[] getPayload() throws MqttException {   return new byte[0];}

综合以上分析可知MQTT的心跳包实际只有2个字节，且第一个字节中前4位表示消息类型后4位表示消息内容。