鸿蒙（HarmonyOS）实现 MQTT 连接（2）

梓墨安

233人浏览 · 2026-10-19 16:04:17

梓墨安 · 2026-10-19 16:04:17 发布

本文是《鸿蒙（HarmonyOS）实现 MQTT 连接：EMQX 服务器搭建 + 客户端开发全流程》的拓展篇。EMQX 服务器创建、ohpm install @ohos/mqtt 安装、module.json5 网络权限配置等基础内容上一篇已经介绍过，这里不再重复，没看过的小伙伴建议先补一下。

成品预览

先看最终效果：顶部是带头像、聊天标题、连接 mqtt 按钮和连接状态灯的标题栏；中间是灰底的消息区——自己的消息靠右显示蓝色气泡，对方的消息靠左显示白色气泡，每个气泡底部带 时:分:秒 时间戳；底部是圆角输入框 + 发送按钮，发送后输入框自动清空。

请添加图片描述

对比上一篇"一个大 TextArea 打印所有日志"的演示界面，本篇主要做了三件事：

升级点	上一篇	本篇
连接方式	1883 非加密端口（自建 Broker）	TLS 直连 EMQX Serverless 的 8883 端口
消息处理	payload 原文直接显示	自定义消息协议，自动区分"自己/对方"
界面	TextArea 日志框	组件化聊天界面（状态栏 + 左右气泡 + 输入栏）

一、连接升级：TLS 直连 EMQX Serverless

上一篇末尾提过一个坑：EMQX Serverless 只开放 8883（MQTT over TLS/SSL）加密端口，直接填 地址:1883 是连不上的。本篇代码补上了这块拼图，改动有两处。

① 连接地址加 ssl:// 前缀，端口改为 8883：

const MQTT_URL = 'ssl://xxxxxxxx.ala.cn-hangzhou.emqxsl.cn:8883'

② connect() 中新增 sslOptions 配置：

await this.client.connect({
  cleanSession: true,
  connectTimeout: 30,
  keepAliveInterval: 60,
  userName: USERNAME,
  password: PASSWORD,
  sslOptions: {
    enableServerCertAuth: false,  // 关闭服务器证书校验
  }
});

enableServerCertAuth: false 表示客户端不校验服务器证书，这样就不用在本地配置 CA 证书文件，学习测试时最省事。

⚠️ 但要清楚：关闭证书校验意味着链路可能被中间人攻击，仅建议学习调试时使用。生产环境应从 EMQX 部署概览页下载 CA 证书并开启校验。

连接成功后把 context 状态置为 true，界面右上角的状态灯就会从红色"未连接"变成绿色"已连接"；连接失败走 catch 分支置回 false：

this.context = true   // 连接成功
...
this.context = false  // 连接失败（catch 分支）

二、设计消息协议：怎么区分"我说的"和"他说的"？

这是本篇最关键的设计点。MQTT 的 payload 只是一个字符串，服务器只负责把消息转发给主题的所有订阅者——它并不知道也不关心消息是谁发的。但聊天界面必须有方向感：自己的消息放右边，别人的消息放左边。

解决办法是自己约定一个消息格式：

客户端ID + "发送:" + 消息内容

例如客户端 ID 为 AAA 的用户发送"你好"，实际发到主题上的 payload 是：

AAA发送:你好

接收方拿到 payload 后分三步解析：

this.first3 = message.payload.substring(0, 3)   // ① 取前 3 个字符（发送者 ID）
this.index = message.payload.indexOf('发送')     // ② 定位"发送"分隔符的位置
// ③ 正文 = payload.substring(index + 3)
//    "发送:"共 3 个字符（发、送、:），从 index+3 开始截取正好跳过分隔符

以 "AAA发送:你好" 为例：前 3 位是 "AAA"；indexOf('发送') 返回 3；正文从下标 6 开始，substring(6) 得到 "你好"。

然后根据前缀判断消息归属：

if (this.first3 != CLIENT_ID) {
  // 不是自己发的 → 对方的消息（num = 1，左侧白色气泡）
} else {
  // 是自己发的 → 自己的消息（num = 2，右侧蓝色气泡）
}

💡 改进提示：substring(0, 3) 是为 3 位客户端 ID（'AAA'）量身定制的，ID 长度一变就会判断失误。更通用的写法是：
const prefix = CLIENT_ID + '发送:'
if (message.payload.startsWith(prefix)) {
// 自己的消息，正文 = message.payload.substring(prefix.length)
} else {
// 对方的消息
}

三、数据模型与状态变量

每条消息要渲染成气泡，需要"内容 + 归属 + 时间"三类信息，用一个 User 接口描述：

interface User{
  name: string   // 消息内容（沿用原代码字段名，注意存的不是"用户名"）
  num: number    // 消息归属：1 = 对方（左侧），2 = 自己（右侧）
  timeh: string  // 时
  timem: string  // 分
  times: string  // 秒
}

组件内的主要状态变量：

@State msg: string = ""           // 调试日志（连接结果等）
@State sendmsg: string = ""       // 输入框内容
@State context: boolean = false   // 连接状态（驱动状态灯变色）
@State index: number = 0          // "发送"分隔符的下标
@State first3: string = ''        // payload 前 3 个字符
@State users: User[] = []         // 消息列表（界面数据源）

users 被声明为 @State，每次 push 新消息都会触发 List 中的 ForEach 重新渲染，新气泡自动追加到列表末尾——这就是消息"一条条冒出来"的原理。

四、界面组件化：@Builder 拆出四个部件

整个聊天界面没有在 build() 里堆成一坨，而是用 @Builder 装饰器拆成 4 个可复用组件，整体骨架是"顶部 / 中部 / 底部"三段式：

build() {
  Column() {
    this.HeaderBar()   // 顶部：标题栏
    Column() {         // 中部：消息列表（layoutWeight(1) 占满剩余高度）
      List() { ForEach(this.users, ...) }
    }.layoutWeight(1)
    this.InputBar()    // 底部：输入栏
  }
}

4.1 HeaderBar —— 顶部标题栏

包含：蓝色圆角头像（Stack + 字母 'A'）、聊天对象名称、"连接 mqtt"按钮、连接状态灯。状态灯用 Circle 画一个小圆点，颜色和文字都由 this.context 三目运算决定：

Row() {
  Circle({ width: 8, height: 8 })
    .fill(this.context ? '#28A745' : '#DC3545')   // 绿=已连接，红=未连接
  Text(this.context ? '已连接' : '未连接')
    .fontSize(12)
    .fontColor(this.context ? '#28A745' : '#DC3545')
    .margin({ left: 6 })
}

标题 Column 设置了 layoutWeight(1) 占据剩余宽度，把按钮和状态灯挤到最右侧。

4.2 TimeBar —— 时间戳组件

一个 Row 里放"时、冒号、分、冒号、秒"五个 Text，字号 8，嵌在气泡底部。原先发送/接收两个气泡里各写了一份时间代码，抽成独立组件后两处复用：

@Builder
TimeBar(content: User) {
  Row() {
    Text(content.timeh).fontSize(8).textAlign(TextAlign.End)
    Text(':').fontSize(8).textAlign(TextAlign.End)
    Text(content.timem).fontSize(8).textAlign(TextAlign.End)
    Text(':').fontSize(8).textAlign(TextAlign.End)
    Text(content.times).fontSize(8).textAlign(TextAlign.End)
  }
}

4.3 SendBubble / ReceiveBubble —— 左右气泡

聊天界面的核心组件，两者结构完全对称，都用 Blank()（自动填满剩余空间）实现左右对齐：

SendBubble（自己，右侧）：Row { Blank(), 气泡 }，蓝色底、白字；
ReceiveBubble（对方，左侧）：Row { 气泡, Blank() }，白色底、深色字。

气泡的"小尾巴"效果靠四个角不同的圆角半径实现——发送气泡 bottomRight: 4，形成指向右下的小尖角；接收气泡相反，bottomLeft: 4：

.borderRadius({
  topLeft: 14,
  topRight: 14,
  bottomLeft: 14,
  bottomRight: 4    // 右下角特别小 → 小尾巴
})

4.4 InputBar —— 底部输入栏

输入框用了 ArkTS 的 $$ 双向绑定语法：

TextInput({ text: $$this.sendmsg })

正向：输入内容自动同步到 this.sendmsg；
反向：代码里把 this.sendmsg 置空，输入框也跟着清空。

所以发送按钮的点击事件里，发完消息一句 this.sendmsg = '' 就能自动清空输入框：

.onClick(async () => {
  await this.sendMessage(`${CLIENT_ID}发送:` + this.sendmsg)
  this.sendmsg = ''
})

五、收发逻辑：一条消息是怎么变成气泡的

5.1 发送

sendMessage() 调用 publish()，把 客户端ID发送:内容 格式的 payload 发到主题。注意一个细节：发送时并不直接把消息塞进 users 数组。

5.2 接收（自己发的也一样）

因为客户端自己也订阅了 testtopic/1，而 MQTT 服务器会把消息转发给包括发送者在内的所有订阅者。所以所有消息——不管谁发的——都统一走 messageArrived() 这一个入口处理：

发送/接收 → messageArrived 回调 → 解析 payload → 生成 User（内容+归属+时间）
        → push 进 users 数组 → ForEach 自动刷新 → 气泡上屏

收发两侧渲染逻辑完全一致，不会出现两条代码路径，这是这个设计比较巧妙的地方。

5.3 时间戳补零

const currentDate = new Date()
const hours = ("0" + currentDate.getHours()).slice(-2)     // 9 → "09"，23 → "023" → "23"
const minutes = ("0" + currentDate.getMinutes()).slice(-2)
const seconds = ("0" + currentDate.getSeconds()).slice(-2)

"0" + 数字 再 slice(-2) 取后两位：个位数补成两位，两位数不受影响，一个很简短的补零技巧。

5.4 归属判断与入列

if (this.first3 != CLIENT_ID) {
  this.users.push({ name: 正文, num: 1, timeh, timem, times })  // 对方 → 左侧
} else {
  this.users.push({ name: 正文, num: 2, timeh, timem, times })  // 自己 → 右侧
}

build() 里用 ForEach 按 num 决定用哪个气泡渲染：

ForEach(this.users, (users: User) => {
  ListItem() {
    if (users.num === 1) {
      this.ReceiveBubble(users)
    } else {
      this.SendBubble(users)
    }
  }
}, (users: User) => users.name)

六、完整代码

import { MqttAsync, MqttConnectOptions, MqttMessage, MQTT_SSL_VERSION } from '@ohos/mqtt';

const MQTT_URL = 'ssl://xxxxxxxx.ala.cn-hangzhou.emqxsl.cn:8883'  // 自己的mqtt的访问地址（EMQX Serverless，TLS端口8883）
const CLIENT_ID = 'AAA' // 客户端ID（需唯一）
const TOPIC = 'testtopic/1'      // 订阅/发布需要的主题
const QOS = 1                   // 服务质量等级（0/1/2）
const USERNAME = '你的用户名'      // 自己登录mqtt时要用的用户名
const PASSWORD = '你的密码'       // 自己登录mqtt时要用的密码

// 消息数据模型：内容 + 归属（1=对方/2=自己）+ 时/分/秒
interface User{
  name:string;
  num:number;
  timeh:string;
  timem:string;
  times:string;
}

@Entry
@Component
struct Index {
  @State msg: string = ""          // 调试日志（连接结果）
  @State sendmsg: string = ""      // 输入框内容
  @State context: boolean = false  // 连接状态（驱动状态灯变色）
  @State index: number = 0         // "发送"分隔符下标
  @State first3:string = ''        // payload 前 3 个字符

  @State users :User[] = [];       // 消息列表（界面数据源）


  //===== 显示时间的组件 =====
  @Builder
  TimeBar(content:User)
  {
    Row(){
      Text(content.timeh)
        .fontSize(8)
        .textAlign(TextAlign.End)
      Text(':')
        .fontSize(8)
        .textAlign(TextAlign.End)
      Text(content.timem)
        .fontSize(8)
        .textAlign(TextAlign.End)
      Text(':')
        .fontSize(8)
        .textAlign(TextAlign.End)
      Text(content.times)
        .fontSize(8)
        .textAlign(TextAlign.End)
    }
  }

  //===== 头部组件 =====
  @Builder
  HeaderBar()
  {
    Column() {
      Row() {
        // 头像
        Stack() {
          Text('A')
            .fontSize(20)
            .fontWeight(FontWeight.Bold)
            .fontColor(Color.White)
        }
        .width(40)
        .height(40)
        .borderRadius(10)
        .backgroundColor('#0D6EFD')

        // 标题
        Column() {
          Text('软件A')
            .fontSize(18)
            .fontWeight(FontWeight.Bold)
            .fontColor('#1F2329')
        }
        .alignItems(HorizontalAlign.Start)
        .margin({ left: 12 })
        .layoutWeight(1)

        Button("连接mqtt")
          .margin({right:50})
          .onClick(() => {this.mqttCommunication();})

        // 连接状态
        Row() {
          Circle({ width: 8, height: 8 })
            .fill(this.context ? '#28A745' : '#DC3545')
          Text(this.context ? '已连接' : '未连接')
            .fontSize(12)
            .fontColor(this.context ? '#28A745' : '#DC3545')
            .margin({ left: 6 })
        }
        .alignItems(VerticalAlign.Center)
      }
      .width('100%')
      .height(60)
      .padding({ left: 16, right: 16 })
      .alignItems(VerticalAlign.Center)

      // 分割线
      Divider()
        .strokeWidth(1)
        .color('#E9ECEF')
    }
    .width('100%')
    .backgroundColor(Color.White)
  }

  // ===== 消息气泡组件 =====
  //发送气泡组件（自己，右侧蓝色）
  @Builder
  SendBubble(content:User)
  {
     Row() {
       Blank()
       Column() {
         Text(content.name)
           .fontSize(14)
           .fontColor(Color.White)
           .lineHeight(20)
           .textAlign(TextAlign.Start)
         this.TimeBar(content)
       }
       .alignItems(HorizontalAlign.End)
       .padding({
         left: 12,
         right: 12,
         top: 10,
         bottom: 8
       })
       .backgroundColor('#0D6EFD')
       .borderRadius({
         topLeft: 14,
         topRight: 14,
         bottomLeft: 14,
         bottomRight: 4
       })
       .border({ width: 1, color: '#E9ECEF' })
     }
     .width('100%')
  }
  //接受气泡组件（对方，左侧白色）
  @Builder
  ReceiveBubble(content:User)
  {
    Row()
    {
      Column() {
        Text(content.name)
          .fontSize(14)
          .fontColor('#1F2329')
          .lineHeight(20)
          .textAlign(TextAlign.Start)
        this.TimeBar(content)
      }
      .alignItems(HorizontalAlign.Start)
      .padding({
        left: 12,
        right: 12,
        top: 10,
        bottom: 8
      })
      .backgroundColor(Color.White)
      .borderRadius({
        topLeft: 14,
        topRight: 14,
        bottomLeft: 4,
        bottomRight: 14
      })
      .border({ width: 1, color: '#E9ECEF' })
      Blank()
    }
    .width('100%')
  }

  // ===== 底部输入栏组件 =====
  @Builder
  InputBar()
  {
      Row() {
        // 输入框（$$ 双向绑定）
        TextInput({ text: $$this.sendmsg })
          .placeholderColor('#ADB5BD')
          .placeholderFont({ size: 14 })
          .fontSize(14)
          .fontColor('#1F2329')
          .backgroundColor('#F1F3F5')
          .borderRadius(20)
          .height(40)
          .layoutWeight(1)
          .padding({ left: 16, right: 16 })
          .onChange((value:string)=>{
            this.sendmsg=value
          })
        // 发送按钮
        Button('发送')
          .fontSize(14)
          .fontColor(Color.White)
          .backgroundColor('#0D6EFD')
          .height(40)
          .width(72)
          .borderRadius(20)
          .margin({ left: 12 })
          .onClick(async ()=>{
            await this.sendMessage(`${CLIENT_ID}发送:`+this.sendmsg)
            this.sendmsg = ''
          })
      }
      .width('100%')
      .padding({ left: 16, right: 16, top: 12, bottom: 12 })
      .alignItems(VerticalAlign.Center)
  }

  
  client = MqttAsync.createMqtt({   //创建异步客户端
    url: MQTT_URL,
    clientId: CLIENT_ID,
    persistenceType: 1  // 客户端持久化类型（0=文件系统，1=内存，2=自定义）
  })
  async mqttCommunication() { //异步连接客户端函数
    try {
      await this.client.connect({//异步访问mqtt地址
        cleanSession: true, // 清除会话（断开后不保留订阅和消息）
        connectTimeout: 30, // 连接超时时间（秒）
        keepAliveInterval: 60, // 心跳间隔（秒），维持长连接
        // 认证信息（若Broker需要）
        userName:USERNAME , // 自己登录mqtt时要用的用户名
        password: PASSWORD, // 自己登录mqtt时要用的密码
        sslOptions:
        {
          enableServerCertAuth: false,  // 关闭服务器证书校验（学习测试用）
        }
      });
      this.msg = 'mqtt： 连接成功'
      this.context = true
      await this.client.subscribe({// 订阅主题
        topic: TOPIC,
        qos: QOS
      });
      this.msg += `已订阅主题：${TOPIC}（QoS： ${QOS}）`

    } catch (err) {
      this.msg += '\n操作失败:' + err.message
      this.context = false
    }
    this.client.messageArrived((TOPIC,message:MqttMessage)=>{//消息到达监听器
      // 解析 payload：前3位=发送者ID，"发送:"后是正文
      this.first3 = message.payload.substring(0,3)
      this.index = message.payload.indexOf('发送')

      // 获取当前时间并补零
      const currentDate = new Date();
      const hours = ("0" + currentDate.getHours()).slice(-2);
      const minutes = ("0" + currentDate.getMinutes()).slice(-2);
      const seconds = ("0" + currentDate.getSeconds()).slice(-2);

      if(this.first3 != CLIENT_ID)
      {
        this.users.push({   // 对方的消息 → 左侧白色气泡
          name: `${message.payload.substring(this.index + 3, message.payload.length)}`,
          num: 1,
          timeh: `${hours}`,
          timem: `${minutes}`,
          times: `${seconds}`
        })
      }
      else
      {
        this.users.push({   // 自己的消息 → 右侧蓝色气泡
          name: `${message.payload.substring(this.index + 3, message.payload.length)}`,
          num: 2,
          timeh: `${hours}`,
          timem: `${minutes}`,
          times: `${seconds}`
        })
      }
    });
  }
  async sendMessage(message:string){//发送消息
    await this.client.publish({
      topic: TOPIC,
      payload: message,
      qos: QOS,
      retained: false, // 是否保留消息（Broker存储最后一条消息）
    });
  }


  build() {
    Column() {
      this.HeaderBar()

      Column() {
          List() {
            ForEach(this.users, (users: User) => {
              ListItem() {
              if (users.num === 1) {
                  this.ReceiveBubble(users)   // 对方 → 左
              } else {
                  this.SendBubble(users)      // 自己 → 右
              }
            }
          }, (users: User) => users.name)
        }.scrollBar(BarState.Off)
      }
      .width('100%')
      .height('100%')
      .layoutWeight(1)
      .backgroundColor('#ffb0b0b0')
      
      this.InputBar()
    }
    .height('100%')
    .width('100%')
  }
}

七、测试方法与注意事项

怎么看到"左右对话"的效果？ 只开一个客户端，永远只能看到自己的右侧蓝气泡，必须制造一个"另一方"：

两台设备：另一台手机/模拟器跑同样的工程，把 CLIENT_ID 改成不同的 3 位值（如 'BBB'），连接同一服务器、同一主题，即可互聊；
电脑模拟：在电脑上用 MQTTX 等 MQTT 客户端工具，用相同的用户名密码连接同一主题，按 XXX发送:内容 的格式发布消息，手机上就会显示左侧白色气泡。

注意事项：

消息归属判断依赖 CLIENT_ID 恰好为 3 位，改 ID 时建议换成第二节的 startsWith 改进写法；
enableServerCertAuth: false 仅用于学习测试，生产环境请下载 CA 证书并开启校验；
ForEach 的键值生成器目前用 users.name（即消息内容），两条内容相同的消息会产生重复键，可能引发渲染异常，更稳妥的写法是 (item, index) => index + item.name；
气泡上的时间是接收方本机时间而非发送时间，两端时钟不一致时会有偏差。