React Native for OpenHarmony 实战：react-native-mqtt 从协议到落地的完整鸿蒙化拆解

这篇记录把 @d11/react-native-mqtt 适配到开源鸿蒙的完整过程。代码在 HarmonyOS 7.0.0(26) 模拟器上跑过，连接、订阅、收包、退订、断开、断线重连七项都是实测的。因为鸿蒙没有现成的 MQTT 客户端库，协议层是从零实现的，所以这篇会把 MQTT 5.0 的报文格式和每一段代码都拆开讲

适配结论速览

• 适配方法：7 个 TurboModule 方法，全部跑在自实现的 MQTT 5.0 协议层之上，协议层用 @ohos.net.socket 起 TCP/TLS
• 关键差异：上游 7 个方法里 6 个走 JSI 全局对象，鸿蒙没有 JSI，改成 TurboModule 统一提供，JS 层调用代码零改动
• 协议工作量：变长字节整数、UTF-8 字符串、二进制、MQTT 5 属性四类原语，加 8 组报文的构建与解析，全部手写
• 验证方式：模拟器连本机自写的最小 MQTT 5.0 broker，七项全通，含断线重连后订阅自动重放
• 产出物：example HAP、独立适配库 HAR（15KB）、中英双版使用文档、验证用 broker 工具

一、为什么第三个库选它

前两篇写了 pulsar 触觉反馈和 file-logger 本地落盘，这篇换一个方向，写设备侧的持续通信

1.1 从 RoboDeck 的实际需求出发

RoboDeck 是机器人调试台，遥操的时候操作员手上只有一个平板。机器人的关节温度、电机电流、电池电压、IMU 数据得持续往回送，一分钟都不能断。这种场景下轮询 HTTP 很笨，延迟高而且浪费连接，长连接才是正路

MQTT 就是干这个的。发布订阅模式，broker 在中间解耦，机器人作为 publisher 只管发，平板作为 subscriber 只管收，双方不需要知道对方的存在。协议本身为弱网设计，有 keepAlive、有 QoS 分级、有会话状态，掉线重连的语义也定义得很清楚

另一个实际好处是 QoS。控制指令走 QoS 1 至少一次，遥测数据走 QoS 0 至多一次，同一个连接上可以按 topic 区分，不需要开两条通道

1.2 库的基本面与选型理由

@d11/react-native-mqtt 在待适配清单里排第 206 位，42 个 star，月下载 405，适配优先级热度 4.24，社区还没有鸿蒙化版本

热度不高，但选它有几个理由。它只有 7 个方法，API 面很小，适合把适配流程完整再走一遍。它的两端上游实现用了不同的 MQTT 客户端库，可以对照着看协议行为。最重要的一点是，鸿蒙侧没有任何 MQTT 客户端库，这意味着协议层要自己写，技术密度比前两个库高得多，能写的东西也多

1.3 先说清楚一个容易混淆的点

npm 上叫 mqtt 的包不止一个。react-native-mqtt（Introvertuous，1.3.1，基于 MQTT 3.1.1）和 @d11/react-native-mqtt（Dream Horizon，0.0.5，基于 MQTT 5.0）是两个完全不同的库。usage-docs 里前者已经有文档了，所以这篇的文档用了 d11-react-native-mqtt.md 避开冲突。后面所有讨论都只针对后者

二、上游怎么实现的

这个库和前面两个都不一样，不先看清架构，适配方向就会错

2.1 混合架构：一半走桥，一半走 JSI

JS 层 src/Modules/mqttModule.ts 在模块被 import 的时候就执行了这么一段：

export const MqttModule: NativeModule & {
  installJSIModule: () => boolean;
} = NativeModules.MqttModule;

declare global {
  var __MqttModuleProxy: MqttModuleProxy;
}

if (!isLoaded()) {
  const mqttModule = NativeModules.MqttModule as {
    installJSIModule: () => boolean;
  };
  const isSuccessful = mqttModule.installJSIModule();
  console.log('::MQTT : JSI bindings installation status', isSuccessful);
}

export const MqttJSIModule = global.__MqttModuleProxy;

它调用原生侧的 installJSIModule()，由原生把 6 个方法挂到 global.__MqttModuleProxy 上。之后 JS 层的方法调用分成两路：

调用方	走的通道
MqttModule.createMqtt(...)	NativeModules.MqttModule，标准桥，返回 Promise
connectMqtt / disconnectMqtt / subscribeMqtt / unsubscribeMqtt / removeMqtt / getConnectionStatusMqtt	global.__MqttModuleProxy，JSI
全部事件	NativeEventEmitter(MqttModule)，标准事件通道

为什么上游要这么设计？看 Android 侧就明白了。MqttModuleImpl.kt 里这 6 个方法确实没有标 @ReactMethod：

override fun removeMqtt(clientId: String) {
  MqttManager.removeMqtt(clientId)
}

override fun getConnectionStatusMqtt(clientId: String): String {
  return MqttManager.getConnectionStatusMqtt(clientId)
}

没有 @ReactMethod 就走不了 RN 的桥，只能通过 JSI 直接调。尤其是 getConnectionStatusMqtt，它要同步返回字符串，桥做不到这件事（桥的方法要么异步带 callback，要么用 isBlockingSynchronousMethod，但那个也有代价）

2.2 事件反而走标准通道

这点对适配是好消息。src/Mqtt/EventEmitter.ts 就是一个标准的 NativeEventEmitter：

function createInstance() {
  return new NativeEventEmitter(MqttModule);
}

Android 侧发事件用的是 DeviceEventManagerModule.RCTDeviceEventEmitter.emit(eventId, payload)，完全是标准套路。所以适配的时候事件链路一行都不用改，要处理的只是那 6 个命令方法

2.3 两端都是 MQTT 5.0

这点我一开始判断错了，以为是 3.1.1，看完两端源码才发现都是 5.0

Android 用 com.hivemq:hivemq-mqtt-client:1.3.0，构建的是 Mqtt5Client：

val client = Mqtt5Client.builder()
  .identifier(clientId)
  .addDisconnectedListener { disconnectedContext -> ... }
  .addConnectedListener { ... }
  .serverHost(host)
  .serverPort(port)

mqtt = if (enableSslConfig) {
  client.sslWithDefaultConfig().buildRx()
} else {
  client.buildRx()
}

连接时用的是 cleanStart(options.cleanSession)，这是 MQTT 5 的 Clean Start 标志位，3.1.1 里叫 Clean Session 但报文格式不同

iOS 用 CocoaMQTT 2.1.5，同样选 5.0：

private let mqtt: CocoaMQTT5
mqtt = CocoaMQTT5(clientID: clientId, host: host, port: UInt16(port))

func mqtt5(_ mqtt5: CocoaMQTT5, didConnectAck ack: CocoaMQTTCONNACKReasonCode, connAckData: MqttDecodeConnAck?) {
  if ack == .success {
    emitJsiEvent(clientId + CONNECTED_EVENT, ["reasonCode": ack.rawValue])
  } else {
    emitJsiEvent(clientId + DISCONNECTED_EVENT, ["reasonCode": ack.rawValue])
  }
}

CocoaMQTTCONNACKReasonCode 是 MQTT 5 的 CONNACK 理由码枚举。JS 侧的 Mqtt5ReasonCode 里那几个值也能对上：134 是用户名或密码错、135 是未授权、140 是认证方法不支持，都是 MQTT 5 规范里的理由码

这个判断直接决定了工作量。 MQTT 3.1.1 的 CONNECT 报文只有协议名、版本、连接标志、keepAlive、clientId、可选的用户名密码，几十行就能写完。MQTT 5 多了整个属性段，CONNECT 有属性、CONNACK 有属性、PUBLISH 有属性、SUBSCRIBE 有属性、DISCONNECT 也有属性，每种属性还有自己的长度编码。工作量差了一个量级

2.4 事件载荷必须逐字复刻

JS 层把这些字段直接透传给业务代码，所以字段名和取值一个都不能差。这是从 Android 源码里逐个抄出来的：

事件名	载荷字段
<clientId>client_initialize	成功 {clientInit: true}；失败 {clientInit: false, errorMessage, errorType: "INITIALIZATION", reasonCode: -6}
<clientId>connected	{reasonCode: 0}
<clientId>disconnected	{reasonCode, errorMessage}
<clientId>mqtt_error	连接失败 {clientConnected: false, errorMessage, errorCause, errorType: "CONNECTION", reasonCode: -2}；退订失败 {clientUnsubscribed: false, errorMessage, errorType: "UNSUBSCRIPTION", topic, reasonCode: -5}
<eventId>subscribe_success	{message, topic, qos}
<eventId>subscribe_failed	{clientSubscribed: false, errorMessage, reasonCode: -4}
<eventId>（消息）	{payload, topic, qos}

负数的理由码是库自己定义的，不是 MQTT 规范的：-1 默认、-2 连接错、-3 断开错、-4 订阅错、-5 退订错、-6 初始化错、-7 响应链错。这些值和 JS 侧 Mqtt5ReasonCode 枚举一一对应，抄错一个业务方的错误处理就会走错分支

2.5 MQTT 5.0 相比 3.1.1 改了什么

既然要自己写协议，先把 5.0 的相对变化理清楚，不然容易按 3.1.1 的印象写错

属性段是整个 5.0 最大的变化。 3.1.1 的报文只有固定字段，要加新功能只能加新报文或者改语义。5.0 给大部分报文加了一个属性段，前面是变长长度，后面是若干条「标识符 + 值」。新特性可以只用新属性承载，不动原有结构，扩展性好了很多。代价就是每种属性的长度编码不同，写错了不会在编译期发现

Clean Session 改叫 Clean Start。 语义也细了：3.1.1 里 Clean Session 同时控制「是否清理会话」和「断开时是否保留会话」，5.0 把它拆成 Clean Start 标志和 Session Expiry Interval 属性两个独立的东西

理由码全面铺开。 3.1.1 只有 CONNACK 和 SUBACK 有理由码，而且 SUBACK 的还是「失败统一给 0x80」。5.0 给 CONNACK、DISCONNECT、PUBACK、PUBREC、PUBREL、PUBCOMP、SUBACK、UNSUBACK、AUTH 都加了理由码，失败原因能精确到「包太大」「配额超了」「主题名不合法」这种程度

流控。 5.0 有 Receive Maximum 属性，双方各自声明能同时处理多少 in-flight 的 QoS 1/2 报文，超了就得等确认。3.1.1 是靠「发到窗口满为止」的隐式机制，不可控

其他还有：主题别名（Topic Alias，长主题可以用两字节的别名代替）、共享订阅（$share/group/topic，多个客户端负载均衡收同一主题）、请求响应模式（Response Topic + Correlation Data）、遗嘱消息也支持属性了

这次适配只实现了属性编解码的框架和用到的几条，主题别名、共享订阅、请求响应、增强认证都没做。这不是偷懒，是权衡：那些特性要用得先有对应的业务场景，而且实现和测试成本都不低，先保证基础链路可用更实际

2.6 本次实现的报文格式

写协议的乐趣在于每个字节都有定义。下面这几张表是这次实际实现的报文格式，字段顺序都是规范里定死的，不能调

CONNECT（客户端发，类型值 1）

部分	字段	长度
固定头	类型 1 + 标志 0	1 字节
固定头	剩余长度	变长
可变字节头	协议名 “MQTT”	2 + 4 字节
可变字节头	协议级别 5	1 字节
可变字节头	连接标志	1 字节
可变字节头	Keep Alive	2 字节
可变字节头	属性	变长长度 + 内容
载荷	Client Identifier	2 + N 字节
载荷	User Name（可选）	2 + N 字节
载荷	Password（可选）	2 + N 字节

PUBLISH（双向，类型值 3）

部分	字段	长度
固定头	类型 3 + 标志（DUP/QoS/RETAIN）	1 字节
可变字节头	主题名	2 + N 字节
可变字节头	报文标识符（QoS > 0 才有）	2 字节
可变字节头	属性	变长长度 + 内容
载荷	应用消息	任意

标志位的排布是第 3 位 RETAIN、第 2-1 位 QoS、第 0 位 DUP。读取的时候要按位掩码取：

const qos = (flags >> 1) & 0x03;
const retain = (flags & 0x01) !== 0;
const dup = (flags & 0x08) !== 0;

SUBSCRIBE（客户端发，类型值 8）

部分	字段	长度
固定头	类型 8 + 标志 0x02	1 字节
可变字节头	报文标识符	2 字节
可变字节头	属性	变长长度 + 内容
载荷	主题过滤器 + 订阅选项	(2 + N) + 1 字节，可重复

标志必须是 0x02，这是规范强制的，写错了服务端会直接断开。订阅选项字节的低 2 位是 QoS，第 2 位 No Local，第 3-4 位 Retain As Published，第 5-6 位 Retain Handling。这次只用了 QoS 那两位

SUBACK（服务端发，类型值 9）

部分	字段	长度
可变字节头	报文标识符	2 字节
可变字节头	属性	变长长度 + 内容
载荷	理由码列表	每条 1 字节

理由码 0x00/0x01/0x02 分别表示授予了 QoS 0/1/2，0x80 及以上是失败。注意授予的 QoS 可能低于订阅时请求的 QoS，比如服务端 Maximum QoS 是 0，你请求 1 它也只会给 0。这次实现是取列表第一条当作授予结果，没有做完整的降级处理

2.7 QoS 的三级语义

MQTT 的 QoS 是应用层设计的，理解它比记住报文流程更重要

QoS 0，至多一次。 发出去就不管了，不确认不重传。报文里没有标识符。丢了就丢了，适合高频遥测数据——丢一帧温度读数无所谓，下一帧马上就来

QoS 1，至少一次。 发送方发 PUBLISH（带标识符），收方回 PUBACK。发送方在合理时间内没收到 PUBACK 就重发，DUP 标志置 1。所以接收方可能收到重复消息，业务侧要么能去重，要么保证重复处理无害。控制指令走这级：宁可多发一次，不能漏发

QoS 2，恰好一次。 四次握手：发送方发 PUBLISH，收方回 PUBREC，发送方发 PUBREL，收方回 PUBCOMP。中间还有报文标识符的占用和释放逻辑。开销大得多，实际项目里用得不多

这次实现了 QoS 0 和 1。QoS 1 的接收侧做了：收到带标识符的 PUBLISH 就回 PUBACK。发送侧的上行 PUBLISH 没实现，因为这个库本身没有 publish 方法。QoS 2 四次握手的 PUBREC/PUBREL/PUBCOMP 完全没有

三、适配方案

3.1 鸿蒙侧的能力边界

先确认手里有什么。@ohos.net.socket 提供：

• constructTCPSocketInstance() 起明文 TCP
• constructTLSSocketInstance(tcpSocket) 在已建立的 TCP 连接上起 TLS
• TCPSocket.connect({address})、send({data})、close()、getState()
• 事件 connect / message / close / error

有 TCP 和 TLS，协议层就能跑。但没有 MQTT 客户端，一个都没有。所以协议层必须自己写

3.2 用 TurboModule 垫片替掉 JSI

完整复刻 JSI 安装机制代价很高，而且收益有限。RNOH 的 TurboModule 体系是现成的，做法是：

1. 鸿蒙侧实现一个标准 TurboModule，提供全部 7 个方法
2. 改 src/Modules/mqttModule.ts，检测到 TurboModule 已带这些方法时就不再装 JSI，直接把 MqttModule 当作 proxy

const mqttModuleFromRegistry = TurboModuleRegistry.get<NativeModule & MqttModuleMethods>(
  'MqttModule'
);

export const MqttModule = (mqttModuleFromRegistry ??
  NativeModules.MqttModule) as NativeModule &
  Partial<MqttModuleMethods> & {installJSIModule?: () => boolean};

if (!isLoaded()) {
  const hasTurboModuleMethods =
    MqttModule !== undefined &&
    MqttModule !== null &&
    typeof MqttModule.connectMqtt === 'function' &&
    typeof MqttModule.subscribeMqtt === 'function';

  if (hasTurboModuleMethods) {
    global.__MqttModuleProxy = MqttModule as unknown as MqttModuleProxy;
  } else if (typeof MqttModule?.installJSIModule === 'function') {
    const isSuccessful = MqttModule.installJSIModule();
    console.log('::MQTT : JSI bindings installation status', isSuccessful);
  } else {
    console.warn(
      '::MQTT : neither JSI bindings nor a TurboModule with MQTT methods are available'
    );
  }
}

这样 Android/iOS 上行为完全不变（那些平台 connectMqtt 不是 @ReactMethod，typeof 检查为 false，走原来的 JSI 分支），鸿蒙上走 TurboModule 分支。JS 层 MqttClient 的调用代码一行都没改

3.3 协议层怎么分层

分成四个文件，职责切开：

文件	职责
MqttProtocol.ts	底层原语：变长字节整数、UTF-8 字符串、二进制、属性。以及 ByteWriter / ByteReader 两个顺序读写器
MqttPackets.ts	各报文的构建与解析，只关心报文格式，不关心连接
MqttConnection.ts	单个客户端连接：socket 管理、拆包、状态机、keepAlive、订阅跟踪与重放
MqttManager.ts	按 clientId 存连接实例，对应上游 Android 的 MqttManager object

MqttTurboModule.ts 在最上面，只做参数校验和事件转发，不含协议逻辑

四、工程化：从 0.72 到 0.84 的迁移

上游 example 工程基于 React Native 0.72.9，而 RNOH 0.84.3 要配 RN 0.84.1。这一步踩的坑比协议层还多，单独拿出来说

4.1 metro 配置整个换掉

上游的 metro.config.js 是 RN 0.72 时代的写法：

const { getDefaultConfig, mergeConfig } = require('@react-native/metro-config');
const escape = require('escape-string-regexp');
const exclusionList = require('metro-config/src/defaults/exclusionList');

metro-config/src/defaults/exclusionList 这个深路径在 0.83 里被 exports 字段禁掉了，直接报 ERR_PACKAGE_PATH_NOT_EXPORTED。RNOH 有自己的配置生成器，换成：

const path = require('path');
const {mergeConfig, getDefaultConfig} = require('@react-native/metro-config');
const {
  createHarmonyMetroConfig,
} = require('@react-native-oh/react-native-harmony/metro.config');

const config = {
  // example 通过相对路径引用仓库根的 src/，需要把仓库根加入 watchFolders
  watchFolders: [path.resolve(__dirname, '..')],
  ...
};

module.exports = mergeConfig(
  getDefaultConfig(__dirname),
  createHarmonyMetroConfig({
    reactNativeHarmonyPackageName: '@react-native-oh/react-native-harmony',
  }),
  config,
);

watchFolders 那行不能少。example 的 src/App.tsx 里是这样引库的：

import { MqttClient } from '../../src/Mqtt/MqttClient';

相对路径指向仓库根，在 example 目录之外。不加 watchFolders metro 根本不看那个目录，打包静默少东西

4.2 包打出来了但库代码不在里面

第一遍打包成功，一检查发现 subscribe_mqtt、client_initialize、getConnectionStatusMqtt 这些字符串一个都不在 bundle 里。按钮文字「Connect Mqtt」在，说明 UI 打包了，库没进去

原因是 metro 从仓库根的 src/ 解析 import ... from 'react-native' 时，去 mqtt-repo/node_modules 找，而依赖全装在 example/node_modules。要在 resolver 里指路：

resolver: {
  // 仓库根没有 node_modules，依赖都装在 example 下，
  // 解析仓库根 src/ 里的 'react-native' 等包时要指到这里
  nodeModulesPaths: [path.resolve(__dirname, 'node_modules')],
  extraNodeModules: {
    'react-native': path.resolve(__dirname, 'node_modules/react-native'),
    react: path.resolve(__dirname, 'node_modules/react'),
  },
},

加完之后再打，库代码就进去了

4.3 babel 配置也要换

上游用 metro-react-native-babel-preset 加 babel-plugin-module-resolver 做别名，RN 0.84 直接换成官方 preset 就行：

module.exports = {
  presets: ['@react-native/babel-preset'],
};

4.4 CLI 版本要钉住

@react-native-oh/react-native-harmony-cli 我按 ^0.84.3 装，结果解析到 0.84.5，和 metro-config 0.83.8 不兼容，还是报 ERR_PACKAGE_PATH_NOT_EXPORTED。钉到 0.84.3 精确版本才好

4.5 app.json 的注册名必须三处一致

第一次装上模拟器屏幕全白。查下来是注册名不一致：app.json 里是 D11MqttExample，而 Index.ets 里写的是 mqtt_example，对不上，RN 找不到要注册的组件

三个地方必须完全相同：

位置	值
example/app.json 的 name	mqtt_example
example/harmony/entry/src/main/ets/pages/Index.ets 的 name	mqtt_example
同一文件的 appKey	mqtt_example

4.6 TurboModule 要 C++ 和 ETS 双侧注册

这是最隐蔽的一个。我一开始觉得 TurboModule 是纯 ArkTS 写的，就把 C++ 侧的注册全删了，只留 ETS。结果运行时报：

Couldn't find Turbo Module on the ArkTs side, name: 'MqttModule'
Suggestions: Have you linked a package that provides this turbo module on the CPP side?

报错信息说得很直白，就是要 CPP 侧也注册。看了 RNOH 的 TurboModuleFactory.cpp 才明白机制：

auto result = this->delegateCreatingTurboModule(ctx, name);
if (result != nullptr) {
    auto arkTSTurboModule = std::dynamic_pointer_cast<const ArkTSTurboModule>(result);
    if (arkTSTurboModule != nullptr && !ctx.arkTSTurboModuleInstanceRef) {
        throw FatalRNOHError(
            std::string("Couldn't find Turbo Module on the ArkTs side, name: '").append(name),
            suggestions);
    }
}

C++ 侧的 delegate 先通过 findArkTSTurboModuleThread 去问 ArkTS 的 TurboModuleProvider 有没有这个模块，有就返回 TaskThread::MAIN 并取得 ArkTS 实例引用。C++ 包不存在，这个查询链条根本起不来

所以要补一套 C++ 侧的文件：

文件	作用
cpp/CMakeLists.txt	编出 librnoh_mqtt.so
cpp/MqttPackage.h	MqttPackage 继承 BaseReactNativeMqttPackage
cpp/generated/RNOH/generated/BaseReactNativeMqttPackage.h	codegen 产物，TurboModule 工厂委托
cpp/generated/RNOH/generated/turbo_modules/MqttModule.h/.cpp	codegen 产物，方法元数据

其中方法元数据要把每个方法的参数个数和是否异步写对：

MqttModule::MqttModule(const ArkTSTurboModule::Context ctx, const std::string name)
    : ArkTSTurboModule(ctx, name) {
    methodMap_ = {
        ARK_ASYNC_METHOD_METADATA(createMqtt, 4),
        ARK_METHOD_METADATA(removeMqtt, 1),
        ARK_METHOD_METADATA(connectMqtt, 2),
        ARK_METHOD_METADATA(disconnectMqtt, 1),
        ARK_METHOD_METADATA(subscribeMqtt, 4),
        ARK_METHOD_METADATA(unsubscribeMqtt, 3),
        ARK_ASYNC_METHOD_METADATA(getConnectionStatusMqtt, 1),
    };
}

ARK_ASYNC_METHOD_METADATA 用于返回 Promise 的方法，ARK_METHOD_METADATA 用于同步 void 的。写反了调用行为会不对

example 侧也要把包注册进 C++：

// PackageProvider.cpp
packages.push_back(std::make_shared<MqttPackage>(ctx));

# CMakeLists.txt
add_subdirectory("${OH_MODULES_DIR}/@react-native-ohos/react-native-mqtt/src/main/cpp" ./mqtt)
target_link_libraries(rnoh_app PUBLIC rnoh_mqtt)

ETS 侧的 PackageProvider.ets 也要注册，两边都不可少：

import {MqttPackage} from '@react-native-ohos/react-native-mqtt/ts';

export function getRNOHPackages(ctx: RNPackageContext): RNOHPackage[] {
  return [
    ...createRNOHPackagesAutolinking(ctx),
    new MqttPackage(ctx),
  ];
}

codegen 产物正常情况下由 hvigor 构建时生成，我这里是手写的，类名要和 TurboModule 名保持一致（MqttModule），我第一版把类名写成 Mqtt 而 Base 包里用的是 Mqtt，反过来又漏改了一处，编译报 use of undeclared identifier 'Mqtt'，统一成 MqttModule 才对

4.7 ArkTS 语言的几个约束

写过 JS 的人转 ArkTS 会被类型系统绊几下，这次碰到的都列出来

UTF-8 编解码要用 util。 没有浏览器那套 TextEncoder 全局对象，得从 @kit.ArkTS 导入：

import {util} from '@kit.ArkTS';

// 编码：string -> Uint8Array
const bytes = new util.TextEncoder().encodeInto(text);

// 解码：Uint8Array -> string
const text = new util.TextDecoder().decodeToString(bytes);

MQTT 的载荷是任意二进制，不保证是合法 UTF-8。上游 Android 用 String(publish.payloadAsBytes) 强转，本身就是有损的，鸿蒙侧用 decodeToString 行为对等

Uint8Array 没有 push。 这是我最开始写得最别扭的地方。JS 里可以 arr.push(byte) 逐字节拼，ArkTS 里不行，会编译报错。所以包了 ByteWriter 用普通数组累积，最后一次性转。如果为了性能想直接用 Uint8Array，得自己维护容量和长度，可读性差很多

Map 遍历要用 for...of 配合解构。 不能直接 map.forEach((v, k) => ...) 吗？其实可以，但 ArkTS 里对回调里的 this 捕获有约束，类方法里写箭头函数访问 this 偶尔会报类型错误。统一用 for (const [k, v] of map) 更稳：

for (const [filter, perTopic] of this.subscriptions) {
  if (!MqttConnection.topicMatches(filter, publish.topic)) continue;
  for (const subscription of perTopic.values()) {
    this.callbacks.onEvent(subscription.eventId, {...});
  }
}

可选链在原生回调里要小心。 this._nativeModule?.addListener(eventType) 这种写法在 RN 的 JS 里是安全的，但 ArkTS 侧如果某个对象可能是 null，?. 之后返回的 undefined 再往下传就容易出类型错误。协议层里我对所有可能为空的对象都做了显式判空，没有依赖可选链

接口和类型不能混用。 ArkTS 里 interface 和 type 定义的对象形状，如果要给 C++ 侧或者跨线程传，得用 Record<string, Object> 这种明确形状。我给事件回调定的签名就是：

export interface MqttConnectionCallbacks {
  onEvent(eventName: string, payload: Record<string, Object>): void;
  onLog(level: 'info' | 'warn' | 'error', message: string): void;
}

用 Record<string, Object> 而不是 any，这样拼装载荷时写错字段名编译期就能发现

4.8 从原生事件到 JS 的完整链路

事件能通不等于理解为什么通。把整条链路走一遍，出问题的时候才知道在哪一段断

JS 侧订阅：

const listener = this.eventEmitter.addListener(eventName, callback);

eventEmitter 是 new NativeEventEmitter(MqttModule)，最终走到 RN 的 RCTDeviceEventEmitter.addListener，在 JS 侧维护一张「事件名 -> 回调列表」的表

原生侧发事件，第一站是 TurboModule：

private emit(eventName: string, payload: Record<string, Object>): void {
  try {
    this.ctx.rnInstance.emitDeviceEvent(eventName, payload);
  } catch (e) {
    const err = e as BusinessError;
    this.logger.warn(`emit ${eventName} failed: ${err.code} ${err.message}`);
  }
}

emitDeviceEvent 是 RNOH 在 AnyThreadRNInstance 上提供的方法。它会经过 C++ 侧的 EventDispatcher，最终调到 JS 的 RCTDeviceEventEmitter.emit，按事件名找到回调列表逐个执行

所以完整链路是：

MqttConnection（协议层收到报文）
  -> MqttTurboModule.emit（组事件名和载荷）
  -> ctx.rnInstance.emitDeviceEvent（RNOH 桥）
  -> C++ EventDispatcher
  -> JS RCTDeviceEventEmitter.emit
  -> EventEmitter.addListener 注册的回调
  -> MqttClient 的业务回调（setOnConnectCallback 等）

这里有个容易混淆的点：MqttConnection 里的 callbacks.onEvent 不是直接发给 JS 的，是先给 MqttTurboModule，由它统一加 clientId 或 eventId 前缀再发。这样协议层不需要知道事件名的拼装规则，只负责在合适的时机触发

协议层触发时机：

协议层动作	触发的事件
客户端创建成功	<clientId>client_initialize，{clientInit: true}
客户端创建失败	<clientId>mqtt_error，errorType: "INITIALIZATION"
收到 CONNACK 且理由码 0	<clientId>connected，{reasonCode: 0}
收到 CONNACK 但理由码非 0	<clientId>mqtt_error，errorType: "CONNECTION"
socket 非主动关闭	<clientId>disconnected
SUBACK 回来且成功	<eventId>subscribe_success
SUBACK 回来但失败	<eventId>subscribe_failed
收到 PUBLISH	<eventId>，带 payload/topic/qos
退订报文发送失败	<clientId>mqtt_error，errorType: "UNSUBSCRIPTION"

主动 disconnect() 不触发任何事件，这是上游语义

4.9 为什么不起线程池

上游 Android 用 Executors.newSingleThreadExecutor() 把所有操作串行化：

object MqttManager {
  private val clientMap: HashMap<String, MqttHelper> = HashMap()
  private val executor: ExecutorService = Executors.newSingleThreadExecutor()

  fun connectMqtt(clientId: String, options: MqttConnectOptions) {
    executor.submit {
      val client = clientMap[clientId]
      if (client != null) client.connect(options)
    }
  }
}

它需要线程池是因为 HiveMQ 在自己的线程上做阻塞 IO，多个客户端实例的操作可能并发，得有人保证同一时刻只有一个在动 socket

鸿蒙侧不需要。@ohos.net.socket 全是异步 API，connect、send 都立即返回，实际 IO 由系统在别的线程做，结果通过事件回调送回。而 ArkTS 的 JS 运行时是单线程的，回调之间天然串行。所以 MqttManager 就是个简单的 Map：

export class MqttManager {
  private static instance: MqttManager | null = null;
  private clients: Map<string, MqttConnection> = new Map();

  static getInstance(): MqttManager {
    if (MqttManager.instance === null) {
      MqttManager.instance = new MqttManager();
    }
    return MqttManager.instance;
  }

  createMqtt(clientId, host, port, enableSslConfig, callbacks): void {
    if (this.clients.has(clientId)) return;   // 已存在则忽略，对齐上游
    this.clients.set(clientId, new MqttConnection(clientId, options, callbacks));
  }
}

少一个线程池，少一类并发问题。但要注意单线程的前提是「所有操作都在 JS 线程发起」，如果哪天把 socket 操作挪到 worker 线程，这个假设就不成立了

五、核心实现拆解

5.1 变长字节整数：MQTT 最特别的编码

MQTT 用变长字节整数（Variable Byte Integer）表示剩余长度和一部分属性值。规则是：每个字节用低 7 位存数据，最高位置 1 表示后面还有字节，置 0 表示这是最后一个。所以 0 到 127 用一个字节，128 到 16383 用两个，最多 4 个字节，上限 268435455

为什么这么设计？因为它让小数值保持紧凑。剩余长度大部分时候小于 128，一个字节就够，省下的带宽在物联网场景里是有意义的

写入：

writeVariableByteInteger(value: number): void {
  if (value < 0 || value > 268435455) {
    throw new MqttProtocolError(`variable byte integer out of range: ${value}`);
  }
  let remaining = value;
  do {
    let byte = remaining % 128;
    remaining = Math.floor(remaining / 128);
    if (remaining > 0) {
      byte = byte | 0x80;
    }
    this.bytes.push(byte);
  } while (remaining > 0);
}

读取：

readVariableByteInteger(): number {
  let multiplier = 1;
  let value = 0;
  let encodedByte = 0;
  do {
    if (this.remaining < 1) {
      throw new MqttProtocolError('unexpected end of packet (vbi)');
    }
    encodedByte = this.data[this.offset++];
    value += (encodedByte & 0x7f) * multiplier;
    if (multiplier > 128 * 128 * 128) {
      throw new MqttProtocolError('malformed variable byte integer');
    }
    multiplier *= 128;
  } while ((encodedByte & 0x80) !== 0);
  return value;
}

注意读取里的 multiplier > 128 * 128 * 128 那个检查。如果第 5 个字节还存在，说明对端发了畸形报文，必须拦住，否则 multiplier 会无限涨下去

5.2 ByteWriter 与 ByteReader

ArkTS 的 Uint8Array 不支持 push，也没法预知长度再填充。所以包了两个类。ByteWriter 内部用普通数组累积，最后一次性转成 Uint8Array：

export class ByteWriter {
  private bytes: number[] = [];

  writeUint8(value: number): void { this.bytes.push(value & 0xff); }
  writeUint16(value: number): void {
    this.bytes.push((value >> 8) & 0xff);
    this.bytes.push(value & 0xff);
  }
  writeBytes(data: Uint8Array): void {
    for (let i = 0; i < data.length; i++) { this.bytes.push(data[i]); }
  }
  toUint8Array(): Uint8Array { return new Uint8Array(this.bytes); }
}

writeUint16 用移位而不是 DataView，纯粹是为了少一个依赖，可读性也更好。ByteReader 反过来，持有一个 Uint8Array 加一个 offset，每个读方法自己推进 offset 并做越界检查

字符串和二进制都是「2 字节长度前缀 + 数据」：

writeUtf8String(value: string): void {
  const encoded = new util.TextEncoder().encodeInto(value);
  if (encoded.length > 65535) {
    throw new MqttProtocolError('utf8 string too long');
  }
  this.writeUint16(encoded.length);
  this.writeBytes(encoded);
}

5.3 属性编解码：我踩的最大的坑

MQTT 5 的属性段前面也是变长长度，后面跟着若干条属性。每条属性第一个字节是属性标识符，后面跟的值长度由标识符决定：

长度	属性
1 字节	Payload Format Indicator、Request Problem Information、Request Response Information、Maximum QoS、Retain Available、Wildcard Subscription Available、Subscription Identifier Available、Shared Subscription Available
2 字节	Server Keep Alive、Topic Alias
4 字节	Receive Maximum、Message Expiry Interval、Session Expiry Interval、Will Delay Interval、Maximum Packet Size
变长整数	Subscription Identifier
UTF-8 字符串	Content Type、Response Topic、Assigned Client Identifier、Authentication Method、Response Information、Server Reference、Reason String
二进制	Correlation Data、Authentication Data
字符串对	User Property

写属性必须按标识符分派：

writeProperties(properties?: MqttProperty[]): void {
  if (properties === undefined || properties.length === 0) {
    this.writeVariableByteInteger(0);
    return;
  }
  const body = new ByteWriter();
  for (const p of properties) {
    body.writeUint8(p.id);
    switch (p.id) {
      // 单字节属性
      case PropertyId.PAYLOAD_FORMAT_INDICATOR:
      case PropertyId.REQUEST_PROBLEM_INFORMATION:
      case PropertyId.REQUEST_RESPONSE_INFORMATION:
      case PropertyId.MAXIMUM_QOS:
        body.writeUint8(p.intValue ?? 0);
        break;
      // 4 字节无符号整数属性
      case PropertyId.RECEIVE_MAXIMUM:
      case PropertyId.SESSION_EXPIRY_INTERVAL:
        body.writeUint32(p.intValue ?? 0);
        break;
      // 2 字节 / 变长 / 字符串 / 二进制各有分支
      ...
    }
  }
  this.writeVariableByteInteger(body.length);
  this.writeBytes(body.toUint8Array());
}

我第一版就漏了单字节那一组。buildConnect 里写了 REQUEST_RESPONSE_INFORMATION（标识符 0x19，十进制 25），运行期直接抛：

MqttProtocolError: unsupported property id: 25

更糟的是这个异常逃出了 socket 回调。onTransportReady 是在 tcpSocket.on('connect') 回调里调的，异常从回调里抛出去，RNOH 没有任何捕获，直接把进程带走：

W AppKit: [MTHRD2364]Record result=0, send event [FRAMEWORK,PROCESS_KILL,JS_ERROR]
W appspawn: com.gewislab.mqtt.example with pid 20164 exit with code:254

应用点一下就没了，hilog 里只有一行 Kill Reason: Js Error

修了两处。一是把单字节属性的分支补全，二是给 socket 回调统一包 try/catch：

private onTransportReady(): void {
  if (this.disposed) return;
  this.clearConnectTimeout();
  try {
    this.send(buildConnect(this.options));
  } catch (e) {
    // 构造报文失败不能让异常逃到 socket 回调外面，否则会直接把应用打挂
    this.handleConnectFailure(`failed to build CONNECT: ${this.describeError(e)}`);
    return;
  }
  this.armKeepAlive();
}

收包那边同样包一层，协议错误降级成断开而不是崩溃

这条经验值得单独记：原生回调里抛异常等于让应用崩溃。 所有注册给系统的回调，要么自己 try/catch，要么确保被调用的代码不可能抛

5.4 CONNECT 报文逐字段构建

固定头第一个字节，高 4 位是报文类型，低 4 位是标志。CONNECT 的类型值是 1，标志固定为 0：

function withFixedHeader(type: number, flags: number, body: Uint8Array): Uint8Array {
  const writer = new ByteWriter();
  writer.writeUint8(((type << 4) | (flags & 0x0f)) & 0xff);
  writer.writeVariableByteInteger(body.length);
  writer.writeBytes(body);
  return writer.toUint8Array();
}

可变字节头的顺序是固定的，规范里写死了：

export function buildConnect(options: ConnectOptions): Uint8Array {
  const body = new ByteWriter();

  // 1. 协议名与协议级别
  body.writeUtf8String(PROTOCOL_NAME);   // "MQTT"
  body.writeUint8(PROTOCOL_VERSION);     // 5

  // 2. 连接标志
  let flags = 0;
  if (options.cleanSession) flags |= CONNECT_FLAG_CLEAN_START;   // 0x02
  if (options.username.length > 0) flags |= CONNECT_FLAG_USER_NAME;  // 0x80
  if (options.password.length > 0) flags |= CONNECT_FLAG_PASSWORD;   // 0x40
  body.writeUint8(flags);

  // 3. Keep Alive，单位秒
  body.writeUint16(options.keepAlive);

  // 4. 属性
  body.writeProperties([
    {id: PropertyId.RECEIVE_MAXIMUM, intValue: 65535},
    {id: PropertyId.REQUEST_RESPONSE_INFORMATION, intValue: 0},
  ]);

  // 5. 载荷：clientId 必选，用户名密码可选
  body.writeUtf8String(options.clientId);
  if (options.username.length > 0) body.writeUtf8String(options.username);
  if (options.password.length > 0) body.writeUtf8String(options.password);

  return withFixedHeader(PacketType.CONNECT, 0, body.toUint8Array());
}

Receive Maximum 声明客户端能同时处理的多大的 in-flight QoS 1/2 消息数，不声明服务端默认是 65535，这里显式写出来。Request Response Information 置 0 表示不需要服务端回响应信息，省点带宽

用户名和密码的标志位要和载荷里的实际内容对得上，只写标志不写内容，服务端会解析错位

5.5 收包拆包：TCP 没有报文边界

TCP 是字节流，发出去的三个报文可能挤在一个 message 事件里到达，也可能一个报文被切成两段。所以必须自己按「固定头 + 变长剩余长度 + 变字节头 + 载荷」的结构切

private drainPackets(): void {
  for (;;) {
    if (this.inboundBuffer.length < 2) return;
    const firstByte = this.inboundBuffer[0];
    const remainingLength = this.readRemainingLength(this.inboundBuffer);
    if (remainingLength === null) return;              // 长度字段还没收全
    const headerSize = this.remainingLengthByteCount(this.inboundBuffer);
    const totalSize = headerSize + remainingLength;
    if (this.inboundBuffer.length < totalSize) return; // 报文还没收全
    const packet = this.inboundBuffer.slice(headerSize, totalSize);
    this.inboundBuffer = this.inboundBuffer.slice(totalSize);

    const type = (firstByte >> 4) & 0x0f;
    const flags = firstByte & 0x0f;
    this.handlePacket(type, flags, packet);
  }
}

三个返回点对应三种「还不能处理」的情况：缓冲区不到 2 字节、长度字段本身没收全、报文主体没收全。剩下已经完整的就切出来处理，切完继续循环，因为缓冲区里可能还有下一个

inboundBuffer 的拼接用 concat，空数组时直接返回另一边，避免无意义的拷贝：

private concat(a: Uint8Array, b: Uint8Array): Uint8Array {
  if (a.length === 0) return b;
  if (b.length === 0) return a;
  const out = new Uint8Array(a.length + b.length);
  out.set(a, 0);
  out.set(b, a.length);
  return out;
}

5.6 收下来的数据是 ArrayBuffer 不是 Uint8Array

这个是查 SDK 声明才发现的。SocketMessageInfo.message 的类型是 ArrayBuffer：

export interface SocketMessageInfo {
  message: ArrayBuffer;
  remoteInfo: SocketRemoteInfo;
}

一开始我直接 const chunk: Uint8Array = info.message;，类型不对。改成包一层：

private onSocketMessage(info: socket.SocketMessageInfo): void {
  if (this.disposed) return;
  // message 事件带回来的是 ArrayBuffer，不是 Uint8Array
  const chunk: Uint8Array = new Uint8Array(info.message);
  this.inboundBuffer = this.concat(this.inboundBuffer, chunk);
  try {
    this.drainPackets();
  } catch (e) {
    this.callbacks.onLog('error', `drainPackets failed: ${this.describeError(e)}`);
    this.handleUnexpectedClose(ReasonCode.RX_CHAIN_ERROR, this.describeError(e));
  }
}

发送那边反过来，TCPSendOptions.data 只接受 string | ArrayBuffer，所以要转：

private send(packet: Uint8Array): void {
  const target = this.tlsSocket !== null ? this.tlsSocket : this.tcpSocket;
  if (target === null) throw new Error('socket is not open');
  target.send({data: MqttConnection.toArrayBuffer(packet)});
}

/** 按实际长度拷一份 ArrayBuffer，避免 view 偏移导致多传字节 */
static toArrayBuffer(data: Uint8Array): ArrayBuffer {
  const out = new ArrayBuffer(data.length);
  new Uint8Array(out).set(data, 0);
  return out;
}

toArrayBuffer 没有直接用 packet.buffer，因为如果 Uint8Array 是某个大 buffer 上的 view，.buffer 会把整个底层 buffer 都传过去。虽然当前代码里所有数组都是精确长度新建的，但封装一层更稳

5.7 主题过滤器通配符匹配

订阅时用的是主题过滤器，可以含 + 和 #。收到消息时不能按 topic 字面量找订阅，要做匹配

MQTT 的规则：主题以 / 分层；+ 匹配单独一层；# 匹配自身所在层及其后所有层，而且必须独占一层，只能放在最后

static topicMatches(filter: string, topic: string): boolean {
  if (filter === topic) return true;
  const filterLevels = filter.split('/');
  const topicLevels = topic.split('/');

  for (let i = 0; i < filterLevels.length; i++) {
    const f = filterLevels[i];
    if (f === '#') {
      // # 必须是最后一层；它匹配剩余的全部层级（含零层）
      return i === filterLevels.length - 1;
    }
    if (i >= topicLevels.length) return false;
    if (f !== '+' && f !== topicLevels[i]) return false;
  }
  return filterLevels.length === topicLevels.length;
}

举个例子。过滤器 robot/+/telemetry 匹配 robot/arm/telemetry，不匹配 robot/arm/joint/telemetry。过滤器 robot/# 匹配 robot、robot/arm、robot/arm/joint/temp 所有以 robot 开头的。过滤器 robot/#/temp 是非法的，# 不在最后

派发的时候遍历全部订阅，用这个函数判断：

private handlePublish(flags: number, packet: Uint8Array): void {
  const publish = parsePublish(flags, packet);
  const payloadText = new util.TextDecoder().decodeToString(publish.payload);

  for (const [filter, perTopic] of this.subscriptions) {
    if (!MqttConnection.topicMatches(filter, publish.topic)) continue;
    for (const subscription of perTopic.values()) {
      this.callbacks.onEvent(subscription.eventId, {
        payload: payloadText,
        topic: publish.topic,
        qos: publish.qos,
      });
    }
  }

  if (publish.qos === 1 && publish.packetId !== undefined) {
    this.send(buildPubAck(publish.packetId, 0));
  }
}

这里有个我第一版写错的地方：一开始我按 publish.topic 直接去 subscriptions 里查，那是字面量查找，通配符订阅永远收不到消息。改成遍历加匹配才对

5.8 订阅跟踪与重连重放

订阅不能只发一个 SUBSCRIBE 报文就不管，得记下来，原因有两个：退订时要判断该主题还有没有别的监听，重连后要重新订阅

用两级 Map：topic -> (eventId -> {qos})

subscribe(eventId: string, topic: string, qos: number): void {
  let perTopic = this.subscriptions.get(topic);
  if (perTopic === undefined) {
    perTopic = new Map<string, Subscription>();
    this.subscriptions.set(topic, perTopic);
  }
  if (perTopic.has(eventId)) {
    this.callbacks.onLog('warn', `duplicate subscribe eventId=${eventId} topic=${topic}`);
    return;
  }
  perTopic.set(eventId, {eventId: eventId, topic: topic, qos: qos});

  if (this.state !== ConnectionState.CONNECTED) {
    // 未连接时先记账，连接成功后 replaySubscriptions 会发出去
    this.callbacks.onLog('info', `subscribe queued until connected: ${topic}`);
    return;
  }
  this.sendSubscribe(eventId, topic, qos);
}

未连接时先记账不发报文，这个很重要。上游 JS 层的用法是 createMqtt 之后可以立刻 subscribe，不一定等连接成功

退订的逻辑：

unsubscribe(eventId: string, topic: string): void {
  const perTopic = this.subscriptions.get(topic);
  if (perTopic === undefined || !perTopic.has(eventId)) {
    this.callbacks.onLog('warn', `unsubscribe for unknown subscription eventId=${eventId} topic=${topic}`);
    return;
  }
  perTopic.delete(eventId);

  if (perTopic.size > 0) {
    // 该 topic 还有其他监听，保留服务端子订阅
    return;
  }
  this.subscriptions.delete(topic);

  if (this.state !== ConnectionState.CONNECTED) return;
  const packetId = this.allocatePacketId();
  this.pendingUnsubscribeIds.set(packetId, {topic: topic, eventId: eventId});
  this.send(buildUnsubscribe(packetId, topic));
}

只有监听数降到 0 才真正发 UNSUBSCRIBE。一个 topic 下多个监听时退掉一个，服务端订阅必须保留，否则其他监听者也收不到了

重连后重放，对齐上游 Android 在 doOnSuccess 里做的事：

private replaySubscriptions(): void {
  for (const perTopic of this.subscriptions.values()) {
    for (const subscription of perTopic.values()) {
      this.callbacks.onLog('info', `replaying subscription ${subscription.topic}`);
      this.sendSubscribe(subscription.eventId, subscription.topic, subscription.qos);
    }
  }
}

这个函数在收到 CONNACK 之后调。实测表现是：broker 重启后我只点了 Connect 没点 Subscribe，broker 仍然收到了 SUBSCRIBE，说明重放生效

5.9 keepAlive 与报文标识符

keepAlive 是服务端和客户端约定的「最长不发报文的时间」。客户端需要在这个间隔内发点东西证明自己还活着，发的是 PINGREQ。实现上取 1.5 倍间隔作为定时器周期，留点余量：

private armKeepAlive(): void {
  this.stopKeepAlive();
  const keepAliveSeconds = this.options.keepAlive > 0 ? this.options.keepAlive : 60;
  const intervalMs = Math.floor(keepAliveSeconds * MqttConnection.KEEP_ALIVE_FACTOR * 1000);
  this.keepAliveTimer = setInterval(() => {
    if (this.state !== ConnectionState.CONNECTED) return;
    try {
      this.send(buildPingReq());
    } catch (e) {
      this.callbacks.onLog('warn', `PINGREQ failed: ${this.describeError(e)}`);
    }
  }, intervalMs);
}

报文标识符（packet id）用于 QoS 大于 0 的报文配对，范围 1 到 65535，不能为 0：

private allocatePacketId(): number {
  if (this.nextPacketId > 65535) this.nextPacketId = 1;
  return this.nextPacketId++;
}

SUBSCRIBE、UNSUBSCRIBE、QoS 1 的 PUBLISH 都要分配。订阅和退订阅还需要用 pendingSubscribeIds / pendingUnsubscribeIds 把 packet id 和业务侧的 eventId 关联起来，这样 SUBACK 回来时才知道该给哪个 eventId 发 subscribe_success

5.10 状态机

三个状态：connecting、connected、disconnected。转换路径：

• connect() 被调用 → connecting，起 socket，发 CONNECT
• 收到 CONNACK 且理由码为 0 → connected，发 connected 事件，重放订阅
• 收到 CONNACK 但理由码非 0 → disconnected，发 mqtt_error
• socket close 或 error → disconnected，发 disconnected 事件
• 主动 disconnect() → 发 DISCONNECT 报文，拆 socket，disconnected，不发 disconnected 事件

最后一条是上游语义：disconnect() 是调用方主动做的，不应该再收到一个「被断开」的通知。用 manualDisconnect 标志区分：

private handleUnexpectedClose(reasonCode: number, message: string): void {
  if (this.disposed || this.manualDisconnect) return;
  this.teardownSocket();
  this.stopKeepAlive();
  this.state = ConnectionState.DISCONNECTED;
  this.callbacks.onEvent(this.clientId + MqttEvent.DISCONNECTED, {
    reasonCode: reasonCode,
    errorMessage: message,
  });
}

六、JS 层改造

6.1 TurboModuleRegistry 而不是 NativeModules

这是第一个坑。我第一版只把 mqttModule.ts 里的 NativeModules.MqttModule 保留着，加了 TurboModule 检测，但 MqttClient.ts 里还写着一句：

const { MqttModule } = NativeModules;

它在 createClient 里用 MqttModule.createMqtt(...)。RNOH 上 NativeModules.MqttModule 拿到的是一个没有方法的存根，createMqtt 是 undefined，调用静默失败

查了 RN 0.84 的 TurboModuleRegistry.js 才明白：

function requireModule(name) {
  if (turboModuleProxy != null) {
    const module = turboModuleProxy(name);
    if (module != null) return module;
  }
  const legacyModule = NativeModules[name];
  if (legacyModule != null) return legacyModule;
  return null;
}

RNOH 的 TurboModule 是经 global.__turboModuleProxy(name) 解析的，NativeModules 是 legacy 回退路径。所以要从 mqttModule.ts 导入：

import { MqttJSIModule, MqttModule } from '../Modules/mqttModule';

改成之后 createMqtt 立刻就能调用了

6.2 同步返回值的兜底

getConnectionStatusMqtt 上游经 JSI 同步返回字符串，TurboModule 只能返回 Promise。JS 层 getConnectionStatus() 拿到 Promise 之后，所有 === CONNECTION_STATE.CONNECTED 的比较都会静默为假——不报错，但逻辑全错

加一个本地缓存：

private nativeStatus: string = CONNECTION_STATE.DISCONNECTED;

getConnectionStatus() {
  const status = MqttJSIModule.getConnectionStatusMqtt(this.clientId);
  if (status instanceof Promise) {
    // 鸿蒙侧拿不到同步返回值：同步返回上次缓存，同时后台刷新缓存。
    status.then((s: string) => { this.nativeStatus = s; }).catch(() => {});
    return this.nativeStatus;
  }
  return status;
}

缓存在 connected 事件和 disconnect() 里也会更新：

// connected 回调里
this.connectionStatus = CONNECTION_STATE.CONNECTED;
this.nativeStatus = CONNECTION_STATE.CONNECTED;

// disconnect() 里
this.connectionStatus = CONNECTION_STATE.DISCONNECTED;
this.nativeStatus = CONNECTION_STATE.DISCONNECTED;

这样调用方拿到的仍然是字符串，行为与上游一致

6.3 一个我自己引入的回归

调试重连的时候发现broker 重启后客户端不重连。查下来是我自己的改动导致的

我当初为了让 getConnectionStatus() 的兜底更准，在断开回调里加了这么一行：

// 断开后把本地镜像状态也置为 disconnected
this.connectionStatus = CONNECTION_STATE.DISCONNECTED;

问题在于上游的 autoReconnect 逻辑就在这个回调里：

this.onDisconnectInterceptor(async (ack) => {
  if (this.connectionStatus === CONNECTION_STATE.CONNECTED) {
    // 指数退避重连
    this.connect(newOptions);
  }
});

我把状态提前置成 DISCONNECTED，这个判断永远为假，重连就静默失效了。改成在 callback 执行完之后再更新，并且加了注释防止以后再犯：

(ack) => {
  // 注意：不要在这里改 connectionStatus。上游的 autoReconnect 逻辑在
  // callback 里判断 connectionStatus === CONNECTED 才决定是否重连，
  // 提前置为 DISCONNECTED 会让重连永远不触发。
  try {
    callback(ack);
  } catch (e) { ... }
}

教训：改别人的逻辑之前，先搞清楚那段逻辑还被谁依赖。 我当时只想着「断开时状态该是 disconnected」，没去想这个回调里还有别的判断

七、跑起来验证

7.1 没有 broker 怎么办

模拟器出不了外网，公共 broker 也用不了。与其去找一个能用的，不如自己写一个。tools/test_broker.py 是个最小 MQTT 5.0 broker，只实现验证需要的部分：CONNECT 回 CONNACK、SUBSCRIBE 回 SUBACK、PUBLISH 转发加 PUBACK、PINGREQ 回 PINGRESP、DISCONNECT 关闭

另外开一个控制端口，可以往里注入消息，这样不需要客户端有 publish 能力也能测收包：

python3 tools/test_broker.py --port 1883 --control-port 1884
printf 'PUB hello hi\n' | nc 127.0.0.1 1884

自己写 broker 还有个好处：broker 侧的日志就是协议的客观证据，比客户端日志更有说服力

7.2 七项实测

example 工程把连接状态、订阅状态、事件流水都显示在界面上，不用连调试器看 console

第一项到第三项，建连、CONNACK、connected 事件：

界面上的事件流水是：

[19:15:11] connected reasonCode=0
connect -> connecting
client created

connect -> connecting 是点击时同步取的状态，connected reasonCode=0 是随后 CONNACK 到达时事件回调写的。两者时间差就是一次网络往返

第四项，broker 从控制端口推一条消息，应用收到：

msg topic=hello qos=0 payload=payload-check-1

topic、qos、payload 三项都对得上。qos 是 0 因为我从控制端口注入时用的是 qos0，符合预期

第五项，退订。broker 侧显示收到 UNSUBSCRIBE，之后再推消息就不再投递了。这一项一开始是失败的，原因在 example 代码而不是协议层——我最初把退订写成「重新 subscribe 一次再 remove」，那样 topic 下会多留一个监听，库判断还有其他监听就不发 UNSUBSCRIBE。改成保存 subscribe 返回的句柄、退订时 remove 同一个句柄就好了

第六项，主动断开。broker 收到 DISCONNECT，界面状态回 disconnected：

broker 日志最后两行就是证据：

[19:25:07] mqttx_d11_rn_mqtt: CONNECT protocol=MQTT/5 keepAlive=60 cleanStart=True
[19:25:07] mqttx_d11_rn_mqtt: SUBSCRIBE hello qos=1   <- 订阅被自动重放

7.3 适配层自己的日志

协议层的运行情况打在 hilog 里。RNOH 的 logger tag 统一是 #RNOH_ARK，clone() 的名字拼在消息体开头，所以搜 MqttLogger 要看消息内容不是 tag：

MqttLogger  created client mqttx_d11_rn_mqtt for 192.168.18.162:1883 ssl=false
MqttLogger  transport ready, sending CONNECT to 192.168.18.162:1883
MqttLogger  CONNACK ok, sessionPresent=false
MqttLogger  replaying subscription hello
MqttLogger  PINGRESP received
MqttLogger  connect failed: connect timeout

最后那行 connect timeout 是 broker 重启期间的第一次重连，30 秒超时后放弃，之后重试成功。这条日志本身也说明了超时兜底在工作

7.4 还没验证的

• TLS 通道。代码路径写了 constructTLSSocketInstance(tcpSocket)，但验证需要带证书的 broker，这次没搭
• QoS 2。参数接受，但协议层只走了 qos0/qos1 的路径，没测
• 长时间运行。数小时连接、切后台再回来的 socket 行为，需要真机

这三项都写进了文档的「功能限制」，没有含糊过去

八、几条经验

1. 先分清上游用 JSI 还是桥，这决定适配是「实现 TurboModule」还是「复刻 JSI 安装」，工作量差一个数量级。判断方法是看原生方法有没有 @ReactMethod，以及 JS 侧是不是从 global 上取方法
2. 属性编解码必须按标识符全部分派。漏一种不会在编译期报错，是在运行期抛，而且是在 socket 回调里抛，直接带走进程
3. 原生回调里抛异常等于崩溃。所有注册给系统的回调要么自己 try/catch，要么保证被调代码不可能抛
4. RNOH 上取 TurboModule 要用 TurboModuleRegistry.get()，NativeModules.X 是 legacy 路径，拿到的是没有方法的存根，而且失败是静默的
5. 同步返回值改成 Promise 之后，所有 === 常量 的比较都会静默失败。这类问题不报错，只是逻辑全错，最难查。要么在 JS 层兜底，要么在文档里写明
6. 改别人的逻辑前先搞清它还被谁依赖。我在断开回调里改了一个状态，结果把上游的重连判断搞失效了，而症状是「不重连」，离改动点很远
7. TCP 拆包要处理三种「还没收全」：缓冲区太短、长度字段没收全、报文主体没收全。少判断一种就会在特定时序下解析错乱
8. 自己写测试 broker 比找公共 broker 值。broker 侧日志是协议的客观证据，而且可以随意制造断线重连场景

九、仓库与文档

资源	地址
适配源码仓库	https://atomgit.com/lhwLHWjackL/d11-react-native-mqtt
上游仓库	https://github.com/dream-horizon-org/d11-react-native-mqtt
使用文档	CPF-RN/usage-docs PR #874（中英双版）

适配仓里 HARMONY_ANALYSIS.md 记了完整的技术分析和验证结果，tools/test_broker.py 是验证用的 broker，harmony/mqtt/src/main/ets/mqtt/ 是协议层四个文件。独立 HAR 构建产物 15KB

待办：真机复验 TLS 通道和长时间运行稳定性，补 QoS 2 或明确写为不支持

---

欢迎加入 CPF-RN 社区：https://atomgit.com/CPF-RN

本文为 AtomGit 开源鸿蒙跨平台技术征文参赛作品，所有代码与截图来自真实适配过程