鸿蒙OS开发实例：【ArkTS 实现MQTT协议】

介绍

MQTT是物联网中的一种协议，在HarmonyOS API9平台，解决方案以C++库移植为实现方案。

遥遥领先的平台，使用MQTT怎能不遥遥领先呢！

新年快乐，本篇将带领你手把手实现HarmonyOS ArkTS语言的MQTT协议。

准备

阅读MQTT 5.0协议
安装Mosquitto服务器
MQTT 联盟中的中国成员实现的PC版客户端参考文档：qr23.cn/AKFP8k

搜狗高速浏览器截图20240326151547.png

效果

HarmonyOS MQTT实现概况

业务流程

页面与功能实现类

MQTT 实现路线

MQTT协议最大的好处是纯软件协议, 不像Zigbee协议还会和硬件相关，所以在实现MQTT协议前，我们只需要专注于两件事情：

MQTT协议传输通道
MQTT协议格式

通道即传输数据的载体，这里指的是Sokect, 了解HarmonyOS平台的Sokect使用方法尤为重要。

数据传输通道打通之后，需要专注于MQTT协议格式的实现。

由于之前对MQTT协议有所了解，所以在HarmonyOS 平台中看到Socket时，觉得在这个新平台上实现MQTT协议理论上没有任何问题，所以最终决定干它。

下边这张图是实现HarmonyOS MQTT协议的过程，仅作参考。

MQTT资深专家，只需要阅读关于HarmonyOS Socket的API 和 TypeScript 操作字节的一些基础内容即可

MQTT协议介绍

一) MQTT共有15种协议

CONNECT
CONNACK
PUBLISH
PUBACK
PUBREC
PUBREL
PUBCOMP
DISCONNECT
PINGREQ
PINGRES
SUBSCRIBE
SUBACK
UNSUBSCRIBE
UNSUBACK
AUTH

二）MQTT共有7种数据格式

这些数据格式名称会在MQTT技术规范中作为标准用语使用

one Byte
Two Byte
Four Byte
UTF-8 String
UTF-8 String Pair
Variable Byte Integer
Binary Data

数据格式实现

数据格式稍微看一下即可

详情可参见MQTT V5.0规范文档

关于数据格式文档可以查找关键词 “1.5 Data representation”

Two Byte

比如数字2，如果要编码为两个字节，则调用twoByte方法即可

twoByte(2)

public static twoByte(content: number): Uint8Array{
   return new Uint8Array([(content & 0xff00) >> 8 , content & 0xff])
}

Four Byte

比如数字2，如果要编码为四个字节，则调用

fourByte(2)

public static fourByte(content: number): Uint8Array{
  return new Uint8Array([(content & 0xff000000) >> 24 ,(content & 0xff0000) >> 16, (content & 0xff00) >> 8 , content & 0xff])
}

UTF-8 String

比如数字2，如果要采用UTF-8编码，则调用

utf8String(2)

public static utf8String(content: string): Uint8Array{
  const encoder = new util.TextEncoder()
  let u8a_encoder = encoder.encodeInto(content)

  let encoderLength = u8a_encoder.length

  let abEncoder = new ArrayBuffer(encoderLength + 2)
  const dv_encoder = new DataView(abEncoder)
  dv_encoder.setInt8(0, (encoderLength & 0xff00) >> 8)
  dv_encoder.setInt8(1, encoderLength & 0x00ff)

  let index: number = 2
  u8a_encoder.forEach( (value) => {
    dv_encoder.setInt8(index++, value)
  })

  return new Uint8Array(abEncoder)
}

UTF-8 String Pair

比如Key为‘Key-Hello’，Value为‘Value-World’如果要采用Key-Value方式的UTF-8编码，则调用

utf8StringPair('Key-Hello','Value-World')

public static utf8StringPair(key: string, value: string): Uint8Array{
  let u8a_key = this.utf8String(key)
  let u8a_value = this.utf8String(value)

  let merge = new Uint8Array(u8a_key.length + u8a_value.length)
  merge.set(u8a_key)
  merge.set(u8a_value, u8a_key.length)
  return merge
}

Variable Byte Integer

可变字节编码，这种编码方式的最大长度是4个字节。

所以，用1～4个字节可动态编码相关数字，实现节省占用内存的目标

比如数字2，编码完成后，只会占用一个字节

encodeVariableByteInteger(2)

public static encodeVariableByteInteger(content: number): Uint8Array {

  let cacheDigit = new Array<number>()

  let numBytes: number = 0
  let no: number = content

  do {
    let digit = no % 128
    no = parseInt(no / 128 +'', )
    if (no > 0) {
      digit |= 0x80
    }
    cacheDigit.push(digit)
    numBytes++
  } while ( (no > 0) && (numBytes < 4) )

  return new Uint8Array(cacheDigit)
}

Binary Data

这种数据格式是由2字节和剩余的字节数组组成，除2个字节之外的字节数组在解码时，需要按照约定来解码。

比如想要表达数字2，则调用

binaryData(new Uint8Array([2]))

public static binaryData(ext: Uint8Array): Uint8Array{
  let length = ext.length

  let u8a_binaryData = new Uint8Array(2 + length)

  u8a_binaryData[0] = (length & 0xff00) >> 8
  u8a_binaryData[1] = length & 0x00ff

  let index: number = 2
  ext.forEach((value)=>{
    u8a_binaryData[index++] = value
  })

  return u8a_binaryData
}

HarmonyOS Socket

[HarmonyOS 指南]

鸿蒙OS开发	更多内容↓点击	HarmonyOS与OpenHarmony技术
鸿蒙技术文档	《鸿蒙NEXT星河版开发学习文档》

1）包引入

注意：这个包在编译时，DevEco Studio 会提示警告，但官方资料中并未做说明

import socket from '@ohos.net.socket';

2）创建Socket

// 创建一个TCPSocket连接，返回一个TCPSocket对象。

let tcp = socket.constructTCPSocketInstance();

3) 注册Socket事件消息

连接事件
接收消息事件
Socket关闭事件
错误事件

tcp.on('message', value => {  
       console.log("on message")  
       let buffer = value.message  
       let dataView = new DataView(buffer)  
       let str = ""  
       for (let i = 0; i < dataView.byteLength; ++i) {    
          str += String.fromCharCode(dataView.getUint8(i))  
       }  
        console.log("on connect received:" + str)
     });
     
tcp.on('connect', () => {  console.log("on connect")});

tcp.on('close', () => {  console.log("on close")});

tcp.on('error', () => {  console.log("on error")});

4) 绑定本地端口

// 绑定IP地址和端口。
let bindAddress = {
  address: '192.168.xx.xx',
  port: 1234, // 绑定端口，如1234
  family: 1
};
tcp.bind(bindAddress, err => {
  if (err) {
    console.log('bind fail');
    return;
  }
  console.log('bind success');
  
  //TODO 连接服务端代码
  ......

});

5) 连接Broker服务器

// 绑定IP地址和端口。
let bindAddress = {
  address: '192.168.xx.xx',
  port: 1234, // 绑定端口，如1234
  family: 1
};
tcp.bind(bindAddress, err => {
  if (err) {
    console.log('bind fail');
    return;
  }
  console.log('bind success');
  
  let connectAddress = {
     address: '192.168.xx.xx',
     port: 5678, // 连接端口，如5678
    family: 1
  };
  tcp.connect({
      address: connectAddress, timeout: 6000
      }, err => {
        if (err) {
         console.log('connect fail');
         return;
        }
      console.log('connect success');
  
      // 发送数据
      tcp.send({
          data: 'Hello, server!'
         }, err => {
           if (err) {
              console.log('send fail');
              return;
            }
          console.log('send success');
       })
   });
});

MQTT 协议实现一瞥

协议的实现过程，需要对照MQTT V5.0规范逐条实现。
切不可操之过急
发布一条消息和接收一条消息用到的协议都是PUBLISH

CONNECT协议

如下代码即为CONNECT协议实现的源码

Socket通道创建成功之后，仅仅代表客户端与服务端已经建立了消息通道，但这条通道是否可以使用，就需要用到“连接“协议，简单点的讲：客户端和服务端开始通信，交流的规则肯定是一致的。

从如下CONNECT协议实现源码注释中可见，一个通信包包含3个部分：1. 固定头（Fixed Header） 2. 可变头（Variable Header）3. 负载体(PayLoad)

注意：固定头表示由两部分组成，第一部分为一个字节，表示数据包类型和其它预留属性，第二部分表示数据包除第一个字节和自己所占用的字节以外，还剩余多少个字节，用于数据包解码参照计算。第二部分的字节长度数据类型为Variable Byte Integer，因此，其可能占用1～4个字节

import ArrayList from '@ohos.util.ArrayList';
import HashMap from '@ohos.util.HashMap';
import util from '@ohos.util';
import MQTTCommon from '../common/MQTTCommon';
import MqttDataTypes from '../common/MqttDataTypes';

//3.1 CONNECT – Connection Request
export default class MQTTConnect{

  //3.1.1 CONNECT Fixed Header 【1字节 + Variable Byte Integer】
  //3.1.2 CONNECT Variable Header
      //3.1.2.1 Protocol Name 【6字节】
      //3.1.2.2 Protocol Version 【1字节】
      //3.1.2.3 Connect Flags 【1字节】
      //3.1.2.10 Keep Alive 【2字节】
      //3.1.2.11 CONNECT Properties
          //3.1.2.11.1 Property Length 【1字节】
          //3.1.2.11.2 Session Expiry Interval 【4字节】
          //3.1.2.11.3 Receive Maximum 【2字节】
          //3.1.2.11.4 Maximum Packet Size 【4字节】
          //3.1.2.11.5 Topic Alias Maximum 【2字节】
          //3.1.2.11.6 Request Response Information 【1字节】
          //3.1.2.11.7 Request Problem Information 【1字节】
          //3.1.2.11.8 User Property【UTF-8 String Pair】
          //3.1.2.11.9 Authentication Method 【UTF-8 String】
          //3.1.2.11.10 Authentication Data【Binary Data】
  //3.1.3 CONNECT Payload
      //3.1.3.1 Client Identifier (ClientID) 【UTF-8 String】
      //3.1.3.2 Will Properties
          //3.1.3.2.1 Property Length 【Variable Byte Integer】
          //3.1.3.2.2 Will Delay Interval 【4字节】
          //3.1.3.2.3 Payload Format Indicator 【1字节】
          //3.1.3.2.4 Message Expiry Interval 【4字节】
          //3.1.3.2.5 Content Type【UTF-8 String】
          //3.1.3.2.6 Response Topic【UTF-8 String】
          //3.1.3.2.7 Correlation Data 【Binary Data】
          //3.1.3.2.8 User Property【UTF-8 String Pair】
     //3.1.3.3 Will Topic 【UTF-8 String】
     //3.1.3.4 Will Payload【Binary Data】
     //3.1.3.5 User Name【UTF-8 String】
     //3.1.3.6 Password【Binary Data】
  //3.1.4 CONNECT Actions

   private connectFlag: number = 0b00000000
   private keepAlive: number = 3600
   private sessionExpiryInterval: number = 3600
   private receiveMaximum: number = 10000
   private topicAliasMaximum: number = 10
   private maximumPacketSize: number = 10000
   private requestResponseInformation: number = 1
   private requestProblemInformation: number = 1
   private clientIdentifier: string = ''
   private authenticationMethod: string = ''
   private authenticationData: string = ''
   private userProperties: HashMap<string, string> = new HashMap()
   private willDelayInterval: number = 0
   private payloadFormatIndicator: number = 1
   private messageExpiryInterval: number = 3600
   private contentType: string = ''
   private responseTopic: string = ''
   private willTopic: string = ''
   private correlationData: string = ''
   private willPayload: string = ''
   private userName: string = ''
   private passWord: string = ''
   private willUserProperties: HashMap<string, string> = new HashMap()

   public static create(): MQTTConnect{
      return new MQTTConnect()
   }

   public setTopicAliasMaximum(topicAliasMaximum: number){
      this.topicAliasMaximum = topicAliasMaximum
      return this
   }

   public setWillUserProperties(willUserProperties: HashMap<string, string>){
      this.willUserProperties = willUserProperties
      return this
   }

   public setUserName(userName: string){
      this.userName = userName
      return this
   }

   public setPassWord(passWord: string){
      this.passWord = passWord
      return this
   }

   public setWillPayload(willPayload: string){
      this.willPayload = willPayload
      return this
   }

   public setCorrelationData(correlationData: string){
      this.correlationData = correlationData
      return this
   }

   public setWillTopic(willTopic: string){
      this.willTopic = willTopic
      return this
   }

   public setResponseTopic(responseTopic: string){
      this.responseTopic = responseTopic
      return this
   }

   public setContentType(contentType: string){
      this.contentType = contentType
      return this
   }

   public setKeepAlive(keepAlive: number){
      this.keepAlive = keepAlive
      return this
   }

   public setSessionExpiryInterval(sessionExpiryInterval: number){
      this.sessionExpiryInterval = sessionExpiryInterval
      return this
   }

   public setReceiveMaximum(receiveMaximum: number){
      this.receiveMaximum = receiveMaximum
      return this
   }

   public setMaximumPacketSize(maximumPacketSize: number){
      this.maximumPacketSize = maximumPacketSize
      return this
   }

   public setRequestResponseInformation(requestResponseInformation: number){
      this.requestResponseInformation = requestResponseInformation
      return this
   }

   public setRequestProblemInformation(requestProblemInformation: number){
      this.requestProblemInformation = requestProblemInformation
      return this
   }

   public setAuthenticationMethod(authenticationMethod: string){
      this.authenticationMethod = authenticationMethod
      return this
   }

   public setAuthenticationData(authenticationData: string){
      this.authenticationData = authenticationData
      return this
   }

   public setUserProperties(m: HashMap<string, string>){
      this.userProperties = m
      return this
   }

   public setClientIdentifier(clientIdentifier: string){
      this.clientIdentifier = clientIdentifier
      return this
   }

   public setWillDelayInterval(willDelayInterval: number){
      this.willDelayInterval = willDelayInterval
      return this
   }

   public setPayloadFormatIndicator(payloadFormatIndicator: number){
      this.payloadFormatIndicator = payloadFormatIndicator
      return this
   }

   public setMessageExpiryInterval(messageExpiryInterval: number){
      this.messageExpiryInterval = messageExpiryInterval
      return this
   }

   //User Name Flag | Password Flag | Will Retain | Will QoS | Will Flag | Clean Start | Reserved
   public setConnectFlag(flag: number){
      this.connectFlag = flag
   }

   public buildPacket(): Uint8Array{
      let allData = new ArrayList<Uint8Array>()

      let remainLength: number = 0

      //3.1.1 CONNECT Fixed Header - 包类型
      let u8a_packettype = MqttDataTypes.oneByte(MQTTCommon.PACKET_TYPE_REQ_CONNECT << 4)
      allData.add(u8a_packettype)

      //3.1.2 CONNECT Variable Header
      //3.1.2.1 Protocol Name
      let u8a_protolName = MqttDataTypes.utf8String('MQTT')
      allData.add(u8a_protolName)
      remainLength += u8a_protolName.length

      //3.1.2.2 Protocol Version
      let u8a_version = MqttDataTypes.oneByte(5)
      allData.add(u8a_version)
      remainLength++

      //3.1.2.3 Connect Flags
      let willFlag = this.connectFlag&0b00001000
       if(willFlag==0){
          //User Name Flag | Password Flag | Will Retain | Will QoS | Will Flag | Clean Start | Reserved
          //bit7             bit6            bit5          bit4~bit3  bit2        bit1           bit0
          //QoS强制设置为0
          //WillRetain强制设置为0
          this.connectFlag = this.connectFlag&0b11000111
       }

      let u8a_connectFlags = MqttDataTypes.oneByte(this.connectFlag)
      allData.add(u8a_connectFlags)
      remainLength += u8a_connectFlags.length

      //3.1.2.10 Keep Alive
      let u8a_keepalive = MqttDataTypes.twoByte(this.keepAlive)
      allData.add(u8a_keepalive)
      remainLength += u8a_keepalive.length

      //3.1.2.11 CONNECT Properties
      let propertiesLength: number = 0
      let insertStartIndex: number = allData.length

      //3.1.2.11.2 Session Expiry Interval
      let u8a_sessionexpiryintervaltype = MqttDataTypes.oneByte(0x11)
      allData.add(u8a_sessionexpiryintervaltype)
      propertiesLength += u8a_sessionexpiryintervaltype.length

      let u8a_sessionexpiryinterval = MqttDataTypes.fourByte(this.sessionExpiryInterval)
      allData.add(u8a_sessionexpiryinterval)
      propertiesLength += u8a_sessionexpiryinterval.length

      //3.1.2.11.3 Receive Maximum
      let u8a_receiveMaximumType = MqttDataTypes.oneByte(0x21)
      allData.add(u8a_receiveMaximumType)
      propertiesLength += u8a_receiveMaximumType.length

      let u8a_receiveMaximum = MqttDataTypes.twoByte(this.receiveMaximum)
      allData.add(u8a_receiveMaximum)
      propertiesLength += u8a_receiveMaximum.length

      //3.1.2.11.4 Maximum Packet Size
      let u8a_maximumPacketSizeType = MqttDataTypes.oneByte(0x27)
      allData.add(u8a_maximumPacketSizeType)
      propertiesLength += u8a_maximumPacketSizeType.length

      let u8a_maximumPacketSize = MqttDataTypes.fourByte(this.maximumPacketSize)
      allData.add(u8a_maximumPacketSize)
      propertiesLength += u8a_maximumPacketSize.length

      //3.1.2.11.5 Topic Alias Maximum
      let u8a_topicAliasMaximumType = MqttDataTypes.oneByte(0x22)
      allData.add(u8a_topicAliasMaximumType)
      propertiesLength += u8a_topicAliasMaximumType.length

      let u8a_topicAliasMaximum = MqttDataTypes.twoByte(this.topicAliasMaximum)
      allData.add(u8a_topicAliasMaximum)
      propertiesLength += u8a_topicAliasMaximum.length

      //3.1.2.11.6 Request Response Information
      let u8a_requestResponseInformationType = MqttDataTypes.oneByte(0x19)
      allData.add(u8a_requestResponseInformationType)
      propertiesLength += u8a_requestResponseInformationType.length

      let u8a_requestResponseInformation = MqttDataTypes.oneByte(this.requestResponseInformation)
      allData.add(u8a_requestResponseInformation)
      propertiesLength += u8a_requestResponseInformation.length

      //3.1.2.11.7 Request Problem Information
      let u8a_requestProblemInformationType = MqttDataTypes.oneByte(0x17)
      allData.add(u8a_requestProblemInformationType)
      propertiesLength += u8a_requestProblemInformationType.length

      let u8a_requestProblemInformation = MqttDataTypes.oneByte(this.requestProblemInformation)
      allData.add(u8a_requestProblemInformation)
      propertiesLength += u8a_requestProblemInformation.length

      //3.1.2.11.8 User Property
      if(this.userProperties && this.userProperties.length > 0){
         //3.3.2.3.7 User Property 【UTF-8 String Pair】
         this.userProperties.forEach( (value, key)=>{
            let a8u_userPropertyType = MqttDataTypes.oneByte(0x26)
            allData.add(a8u_userPropertyType)
            propertiesLength += a8u_userPropertyType.length

            let a8u_userProperty = MqttDataTypes.utf8StringPair(key, value)
            allData.add(a8u_userProperty)
            propertiesLength += a8u_userProperty.length
         })
      }

      //3.1.2.11.9 Authentication Method
      if(this.authenticationMethod && this.authenticationMethod.length > 0){
         let u8a_authenticatinMethodType = MqttDataTypes.oneByte(0x15)
         allData.add(u8a_authenticatinMethodType)
         propertiesLength += u8a_authenticatinMethodType.length

         let u8a_authenticatinMethod = MqttDataTypes.utf8String(this.authenticationMethod)
         allData.add(u8a_authenticatinMethod)
         propertiesLength += u8a_authenticatinMethod.length
      }

      //3.1.2.11.10 Authentication Data
      if(this.authenticationData && this.authenticationData.length > 0){
         let u8a_authenticatinDataType = MqttDataTypes.oneByte(0x16)
         allData.add(u8a_authenticatinDataType)
         propertiesLength += u8a_authenticatinDataType.length

         let targetString: string = this.authenticationData
         let textEncoder = new util.TextEncoder()
         let binaryArray = textEncoder.encodeInto(targetString);

         let u8a_authenticatinData = MqttDataTypes.binaryData(binaryArray)
         allData.add(u8a_authenticatinData)
         propertiesLength += u8a_authenticatinData.length
      }

      //3.1.2.11.1 Property Length
      let u8a_propertylength = MqttDataTypes.encodeVariableByteInteger(propertiesLength)
      allData.insert(u8a_propertylength, insertStartIndex)

      remainLength += (u8a_propertylength.length  + propertiesLength)

      //3.1.3 CONNECT Payload
      //3.1.3.1 Client Identifier (ClientID)
      if(this.clientIdentifier == null || this.clientIdentifier.length == 0){
         this.clientIdentifier = "MQTT-" + new Date().getTime() + '-' + Math.random()
      }
      let u8a_clientidentifier = MqttDataTypes.utf8String(this.clientIdentifier)
      allData.add(u8a_clientidentifier)
      remainLength += u8a_clientidentifier.length

      //3.1.3.2 Will Properties
      if(willFlag == 1){
         let willPropertiesLength: number = 0
         let willPropertiesIndex: number = allData.length

         //3.1.3.2.2 Will Delay Interval
         //seconds
         let u8a_willDelayIntervalType = MqttDataTypes.oneByte(0x18)
         allData.add(u8a_willDelayIntervalType)
         willPropertiesLength += u8a_willDelayIntervalType.length

         let u8a_willDelayInterval = MqttDataTypes.fourByte(this.willDelayInterval)
         allData.add(u8a_willDelayInterval)
         willPropertiesLength += u8a_willDelayInterval.length

         //3.1.3.2.3 Payload Format Indicator
         let u8a_payloadformatIndicatorType = MqttDataTypes.oneByte(0x18)
         allData.add(u8a_payloadformatIndicatorType)
         willPropertiesLength += u8a_payloadformatIndicatorType.length

         let u8a_payloadformatIndicator = MqttDataTypes.fourByte(this.payloadFormatIndicator)
         allData.add(u8a_payloadformatIndicator)
         willPropertiesLength += u8a_payloadformatIndicator.length

         //3.1.3.2.4 Message Expiry Interval
         let u8a_messageExpiryIntervalType = MqttDataTypes.oneByte(0x02)
         allData.add(u8a_messageExpiryIntervalType)
         willPropertiesLength += u8a_messageExpiryIntervalType.length

         let u8a_messageExpiryInterval = MqttDataTypes.fourByte(this.messageExpiryInterval)
         allData.add(u8a_messageExpiryInterval)
         willPropertiesLength += u8a_messageExpiryInterval.length

         //3.1.3.2.5 Content Type
         let u8a_contentType = MqttDataTypes.oneByte(0x03)
         allData.add(u8a_contentType)
         willPropertiesLength += u8a_contentType.length

         let u8a_content = MqttDataTypes.utf8String(this.contentType)
         allData.add(u8a_content)
         willPropertiesLength += u8a_content.length

         //3.1.3.2.6 Response Topic
         let u8a_responseTopicType = MqttDataTypes.oneByte(0x08)
         allData.add(u8a_responseTopicType)
         willPropertiesLength += u8a_responseTopicType.length

         let u8a_responseTopic = MqttDataTypes.utf8String(this.responseTopic)
         allData.add(u8a_responseTopic)
         willPropertiesLength += u8a_responseTopic.length

         //3.1.3.2.7 Correlation Data
         let u8a_correlationDataType = MqttDataTypes.oneByte(0x09)
         allData.add(u8a_correlationDataType)
         willPropertiesLength += u8a_correlationDataType.length

         let targetString: string = this.correlationData
         let textEncoder = new util.TextEncoder()
         let binaryArray = textEncoder.encodeInto(targetString);

         let u8a_correlationData = MqttDataTypes.binaryData(binaryArray)
         allData.add(u8a_correlationData)
         willPropertiesLength += u8a_correlationData.length

         //3.1.3.2.8 User Property
         if(this.willUserProperties && this.willUserProperties.length > 0){
            //3.3.2.3.7 User Property 【UTF-8 String Pair】
            this.willUserProperties.forEach( (value, key)=>{
               let a8u_userPropertyType = MqttDataTypes.oneByte(0x26)
               allData.add(a8u_userPropertyType)
               willPropertiesLength += a8u_userPropertyType.length

               let a8u_userProperty = MqttDataTypes.utf8StringPair(key, value)
               allData.add(a8u_userProperty)
               willPropertiesLength += a8u_userProperty.length
            })
         }

         //3.1.3.2.1 Property Length
         let u8a_propertylength = MqttDataTypes.encodeVariableByteInteger(willPropertiesLength)
         allData.insert(u8a_propertylength, willPropertiesIndex)

         remainLength += (u8a_propertylength.length  + willPropertiesLength)

      }

      //3.1.3.3 Will Topic
      if(willFlag == 1){
         let u8a_willTopic = MqttDataTypes.utf8String(this.willTopic)
         allData.add(u8a_willTopic)
         remainLength += u8a_willTopic.length
      }

      //3.1.3.4 Will Payload
      if(willFlag == 1){
         let targetString: string = this.willPayload
         let textEncoder = new util.TextEncoder()
         let binaryArray = textEncoder.encodeInto(targetString);

         let u8a_willPayload = MqttDataTypes.binaryData(binaryArray)
         allData.add(u8a_willPayload)
         remainLength += u8a_willPayload.length
      }

      //3.1.3.5 User Name
      if(willFlag == 1){
         if(((this.connectFlag&0b10000000) >> 7) == 1){
            let u8a_willTopic = MqttDataTypes.utf8String(this.userName)
            allData.add(u8a_willTopic)
            remainLength += u8a_willTopic.length
         }
      }

      //3.1.3.6 Password
      if(willFlag == 1){
         if(((this.connectFlag&0b01000000) >> 6) == 1){
            let targetString: string = this.passWord
            let textEncoder = new util.TextEncoder()
            let binaryArray = textEncoder.encodeInto(targetString);

            let u8a_willPayload = MqttDataTypes.binaryData(binaryArray)
            allData.add(u8a_willPayload)
            remainLength += u8a_willPayload.length
         }
      }

      //3.1.1 CONNECT Fixed Header - 包剩余长度
      let u8a_remainlength = MqttDataTypes.encodeVariableByteInteger(remainLength)
      allData.insert(u8a_remainlength, 1)

      //合并数据
      let allUint8Array = new Uint8Array(1 + u8a_remainlength.length + remainLength)

      let allUint8ArrayIndex: number = 0

      allData.forEach((element: Uint8Array)=>{
           element.forEach( (value) => {
              allUint8Array[allUint8ArrayIndex++] = value
           })
      })

      return allUint8Array

   }

}

最后呢，很多开发朋友不知道需要学习那些鸿蒙技术？鸿蒙开发岗位需要掌握那些核心技术点？为此鸿蒙的开发学习必须要系统性的进行。

而网上有关鸿蒙的开发资料非常的少，假如你想学好鸿蒙的应用开发与系统底层开发。你可以参考这份资料，少走很多弯路，节省没必要的麻烦。由两位前阿里高级研发工程师联合打造的《鸿蒙NEXT星河版OpenHarmony开发文档》里面内容包含了（ArkTS、ArkUI开发组件、Stage模型、多端部署、分布式应用开发、音频、视频、WebGL、OpenHarmony多媒体技术、Napi组件、OpenHarmony内核、Harmony南向开发、鸿蒙项目实战等等）鸿蒙（Harmony NEXT）技术知识点

如果你是一名Android、Java、前端等等开发人员，想要转入鸿蒙方向发展。可以直接领取这份资料辅助你的学习。下面是鸿蒙开发的学习路线图。