长连接通道

更新时间：2026-10-27 19:49:17

长连接通道（MQTT）提供 IoT 业务协议封装的云端数据下行能力，为 App 提供设备 Topic 的订阅、取消订阅、发布消息（无业务回执）、云端下行数据监听和连接状态监听等能力。

MQTT 连接由 SDK 按设备所属 region 自动建立并复用，业务无需传入服务器地址。

// 导入长连接通道
import { RTCXMqttService, RTCXMqttType } from 'rtcxsdk'

接口定义

export type LinkAppCompleteCallback = (error: BusinessError<void> | undefined) => void

export interface IRTCXLinkAppService {
  /**
   * 当前连接状态
   */
  readonly status: RTCXLinkAppStatus

  /**
   * 添加下行数据监听
   */
  addLinkDownDataListener(listener: IRTCXLinkAppDownListener): void

  /**
   * 移除下行数据监听
   */
  removeLinkDownDataListener(listener: IRTCXLinkAppDownListener): void

  /**
   * 订阅设备Topic
   *
   * @param topic 预留参数，建议传 undefined
   * @param productKey 设备型号
   * @param deviceName 设备序列号
   * @param iotId 设备唯一标识
   * @param deviceType 设备类型
   * @param complete 订阅结果回调
   */
  subscribeToTopic(topic: string | undefined, productKey: string, deviceName: string, iotId: string,
    deviceType: MqttDeviceType, complete?: LinkAppCompleteCallback): void

  /**
   * 取消订阅设备Topic
   */
  unsubscribeToTopic(topic: string | undefined, productKey: string, deviceName: string, iotId: string,
    deviceType: MqttDeviceType, complete?: LinkAppCompleteCallback): void

  /**
   * 发布消息，无业务回执
   */
  publishToTopic(topic: string | undefined, productKey: string, deviceName: string, iotId: string,
    deviceType: MqttDeviceType, data: Uint8Array, complete?: LinkAppCompleteCallback): void

  /**
   * 添加连接状态监听
   */
  addLinkConnectListener(listener: ILRTCXLinkAppConnectListener): void

  /**
   * 移除连接状态监听
   */
  removeLinkConnectListener(listener: ILRTCXLinkAppConnectListener): void

  /**
   * 建立连接
   */
  connect(): void

  /**
   * 断开所有连接
   */
  disconnect(): void

  /**
   * 重连所有连接
   */
  reconnect(): void
}

数据结构

// 连接状态
export enum RTCXLinkAppStatus {
  // 已连接
  Connected = 0,
  // 连接被拒绝
  Refused = 1,
  // 连接已关闭
  Closed = 2,
  // 连接异常
  Error = 3,
  // 协议错误
  ProtocolError = 4,
  // 被服务端关闭
  ClosedByBroker = 5
}

// 设备类型
export enum MqttDeviceType {
  // 普通设备
  Normal = 0,
  // 专属版设备
  Private = 1,
  // AP设备
  Ap = 2
}

// 连接状态监听者
export interface ILRTCXLinkAppConnectListener {
  /**
   * 通道连接状况改变
   * @param connectState 参见 RTCXLinkAppStatus
   */
  onLinkAppConnectState: (connectState: RTCXLinkAppStatus) => void
}

// 下行数据监听者
export interface IRTCXLinkAppDownListener {
  /**
   * 下行数据回调
   * @param data 消息内容
   * @param topic 下推Topic的业务部分
   * @param iotId 设备唯一标识
   */
  onLinkDownData: (data: Uint8Array | undefined, topic: string | undefined, iotId: string) => void

  /**
   * 上抛前的过滤回调，返回 false 则不上抛
   */
  arrowHandle?: (iotId: string, topic: string | undefined) => boolean
}

// MQTT错误码
export enum EMqttError {
  // 取消订阅时尚未订阅
  SUBSCRIBE_REQUIRED = -1,
  // 当前连接已断开
  SESSION_DISCONNECT = -2,
  // 无法获取MQTT通道
  CLIENT_UNAVAILABLE = -3,
  // 服务端无响应
  NO_RESPONSE = -4,
  // 连接超时
  CONNECT_TIMEOUT = -5,
  // 协议错误
  PROTOCOL_ERROR = -6,
  // 订阅失败
  SUBSCRIBE_FAILED = -7
}

订阅

长连接通道采用 MQTT 协议，基于订阅/发布模型设计。订阅设备 Topic 后，当云端或设备向该 Topic 发布消息时，可通过下行监听收到消息。

RTCXMqttService.subscribeToTopic(undefined,
  deviceInfo.productKey,
  deviceInfo.deviceName,
  deviceInfo.iotId,
  RTCXMqttType.MqttDeviceType.Normal,
  (error: BusinessError<void> | undefined) => {
    if (error) {
      showToast(`订阅失败：${error.code} ${error.message}`)
      return
    }
    // 订阅成功
  })

备注

• topic 为预留参数，业务无需拼接 Topic，传 undefined 即可；当前仅支持设备维度订阅，不支持自定义 Topic。
• 普通设备传 MqttDeviceType.Normal（0），专属版设备传 Private（1），AP 设备传 Ap（2）。
• 重复订阅同一台设备时会直接返回成功，不会重复发送订阅请求。
• 连接尚未建立时，订阅会被缓存，待连接成功后由 SDK 自动完成订阅；首次订阅的回调会在订阅完成后触发，业务可自行增加超时处理。
• 设备所属 region 查询失败时返回 CLIENT_UNAVAILABLE（-3），订阅被服务端拒绝时返回 SUBSCRIBE_FAILED（-7），连接中断导致请求无响应时返回 NO_RESPONSE（-4）。

取消订阅

取消订阅是订阅的逆过程，取消后将不再收到该设备的云端下行消息。

RTCXMqttService.unsubscribeToTopic(undefined,
  deviceInfo.productKey,
  deviceInfo.deviceName,
  deviceInfo.iotId,
  RTCXMqttType.MqttDeviceType.Normal,
  (error: BusinessError<void> | undefined) => {
    if (error) {
      showToast(`取消订阅失败：${error.code} ${error.message}`)
      return
    }
    // 取消订阅成功
  })

备注

• 未订阅时调用取消订阅返回 SUBSCRIBE_REQUIRED（-1）。
• 当前连接已断开时返回 SESSION_DISCONNECT（-2）。
• 调用 disconnect() 会关闭连接并清空订阅缓存，无需逐个取消订阅。

发布消息

向设备 Topic 发布消息。该接口没有业务回执，回调仅表示报文已提交发送。

import { util } from '@kit.ArkTS'

const payload: Uint8Array = util.TextEncoder.create().encodeInto(JSON.stringify({
  // 业务参数
}))

RTCXMqttService.publishToTopic(undefined,
  deviceInfo.productKey,
  deviceInfo.deviceName,
  deviceInfo.iotId,
  RTCXMqttType.MqttDeviceType.Normal,
  payload,
  (error: BusinessError<void> | undefined) => {
    if (error) {
      showToast(`发布失败：${error.code} ${error.message}`)
      return
    }
    // 消息已发送
  })

备注

• data 为 Uint8Array，业务 JSON 需先编码为 UTF-8 字节数组。
• 发布使用 QoS 0，没有服务端业务回执，回调成功仅代表报文已通过连接发送。
• 连接已断开时返回 SESSION_DISCONNECT（-2）。

注册下行数据监听

注册下行监听后，才能收到已订阅设备的云端下行数据。建议在发起订阅前注册。

import { util } from '@kit.ArkTS'

const downListener: RTCXMqttType.IRTCXLinkAppDownListener = {
  // 可选：返回 false 时SDK不再回调该监听者的 onLinkDownData
  arrowHandle: (iotId: string, topic: string | undefined): boolean => {
    return iotId === deviceInfo.iotId
  },
  onLinkDownData: (data: Uint8Array | undefined, topic: string | undefined, iotId: string) => {
    if (!data || data.length === 0) {
      return
    }
    const text = util.TextDecoder.create('utf-8').decodeToString(data)
    console.info(`mqtt down data, iotId: ${iotId}, topic: ${topic}, data: ${text}`)
  }
}

// 注册监听
RTCXMqttService.addLinkDownDataListener(downListener)

// 不需要时移除监听
// RTCXMqttService.removeLinkDownDataListener(downListener)

备注

当前版本回调中的 topic 参数固定为 undefined，可使用 iotId 过滤感兴趣的消息；data 为设备下行原始数据，格式由具体业务定义。

注册连接状态监听

注册通道连接状态监听后，连接状态变化时会收到通知。

const connectListener: RTCXMqttType.ILRTCXLinkAppConnectListener = {
  onLinkAppConnectState: (connectState: RTCXMqttType.RTCXLinkAppStatus) => {
    console.info(`mqtt connect state: ${connectState}`)
  }
}

// 注册监听
RTCXMqttService.addLinkConnectListener(connectListener)

// 不需要时移除监听
// RTCXMqttService.removeLinkConnectListener(connectListener)

连接状态说明：

状态	值	说明
Connected	0	已连接
Refused	1	连接被拒绝（鉴权失败等）
Closed	2	连接已关闭，连接关闭后统一回调该状态
Error	3	连接异常或连接超时
ProtocolError	4	协议错误
ClosedByBroker	5	被服务端关闭

备注

• status 为 SDK 维度的连接状态。多 region 场景下可能存在多条连接，状态以最后一次变化为准。
• 连接异常关闭时，会先回调具体原因（如 Error、ClosedByBroker），再回调 Closed；SDK 随后按重连策略自动重连。

连接与重连

连接由订阅动作按需建立，业务通常无需主动调用 connect()。

// 当前连接状态
const currentStatus: RTCXMqttType.RTCXLinkAppStatus = RTCXMqttService.status

// 手动重连所有连接
RTCXMqttService.reconnect()

// 断开所有连接并清空订阅缓存
RTCXMqttService.disconnect()

备注

• connect() 当前为空实现，连接会在首次订阅时自动建立；无需在初始化后主动调用。
• 首次订阅时，若 SDK 未缓存设备所属 region，会先查询 region（超时 3 秒），再按 region 建立或复用 TLS 连接，单次连接超时 30 秒。
• 连接意外断开后，SDK 每隔 10 秒自动重连，最多重试 12 次；重连成功后会恢复已缓存的订阅。
• 自动重连次数耗尽后，可调用 reconnect() 手动重连。调用 disconnect() 会停止自动重连并清空订阅缓存，再次使用需重新订阅。

错误码

错误码	值	说明
SUBSCRIBE_REQUIRED	-1	取消订阅时尚未订阅
SESSION_DISCONNECT	-2	当前连接已断开
CLIENT_UNAVAILABLE	-3	无法获取 MQTT 通道（如 region 或服务地址获取失败）
NO_RESPONSE	-4	请求等待期间连接中断，服务端无响应
CONNECT_TIMEOUT	-5	连接超时
PROTOCOL_ERROR	-6	协议错误
SUBSCRIBE_FAILED	-7	订阅被服务端拒绝

备注

当前版本订阅、取消订阅、发布回调中实际可能返回 -1、-2、-3、-4、-7；连接超时和协议错误通过连接状态 Error、ProtocolError 反馈。

注意事项

• 使用前需完成 SDK 初始化并登录，MQTT 鉴权依赖登录态。
• 建议在页面显示时注册下行数据和连接状态监听，页面销毁时调用 removeLinkDownDataListener、removeLinkConnectListener 移除监听，避免回调泄漏。
• 只需订阅当前业务涉及的设备，长期不使用的设备建议取消订阅。
• 下行数据为原始字节流，需按业务协议解析，避免在主线程执行耗时处理。