长连接通道

更新时间：2026-08-18 19:22:43

长连接通道（MQTT），提供IoT业务协议封装的云端数据下行能力，flutter SDK同时支持iOS端和Android端，为App提供订阅、取消订阅、发布（暂未启用）、云端下行消息监听等能力。

// 导入头文件
import 'package:airtc_engine/src/app_express/rtcx_mqtt_client_controller.dart';

API接口

长连接通道相关API接口如下：

class RTCXFluMQTTClientController {
  /// 订阅设备消息
  @async
  RTCXFluMQTTClientResult subscribe(RTCXFluMQTTClientConfig config);

  /// 取消订阅
  @async
  RTCXFluMQTTClientResult unsubscribe(RTCXFluMQTTClientConfig config);

  /// 设置订阅消息监听
  void addOnMessageListener(RTCXFluMQTTClientConfig config);

  /// 移除订阅消息监听
  void removeOnMessageListener(RTCXFluMQTTClientConfig config);

  /// 释放资源
  void destroy();
}

abstract class RTCXFluMqttClientCallback {
  /// 客户端是否已连接
  void onConnectState(bool isConnected);

  /// 接收订阅消息
  void onMessageArrived(String topic, String message);
}

初始化API

  // 使用RTCXFluMQTTClientController，初始化时候需要传入config，config需要设备信息
  RTCXFluMQTTClientController? _controller;
  bool? _connected;
  
  @override
  void initState() {
    super.initState();
    _controller = RTCXFluMQTTClientController(
      config: RTCXFluMQTTClientConfig(
        productKey: widget.productKey,
        deviceName: widget.deviceName,
        iotId: widget.iotId,
        deviceType: widget.deviceType,
      ),
    );
    _controller!.onConnectState.listen((bool isConnected) {
      if (mounted) {
        setState(() => _connected = isConnected);
      }
    });
    _controller!.onMessageArrived.listen((RTCXFluMQTTClientMessage message) {
      if (mounted) {
        print("onMessageArrived topic:${message.topic}, message:${message.message}");
      }
    });
  }

订阅

  // 长连接通道，采用MQTT协议，基于订阅/发布模型设计。
  Future<void> _subscribe() async {
    try {
      final result = await _controller!.subscribe();
      await _controller!.addOnMessageListener();
      setState(() {
        _subscribed = result.result;
      });
      Fluttertoast.showToast(msg: result.result ? '订阅成功' : '订阅失败');
    } catch (e) {
      Fluttertoast.showToast(msg: '订阅失败: $e');
    }
  }

取消订阅

  Future<void> _unsubscribe() async {
    try {
      final result = await _controller!.unsubscribe();
      setState(() {
        _unsubscribed = result.result;
      });
      Fluttertoast.showToast(msg: '已取消订阅');
    } catch (e) {
      Fluttertoast.showToast(msg: '取消订阅失败: $e');
    }
  }

设置订阅消息回调

    // 可以在initState中，_controller初始化后进行监听，监听收到的消息
    _controller!.onMessageArrived.listen((RTCXFluMQTTClientMessage message) {
      if (mounted) {
        print("onMessageArrived topic:${message.topic}, message:${message.message}");
      }
    });

设置状态监听回调

监听连接状态，当连接状态发生变化时，可接收到状态通知

    // 可以在initState中，_controller初始化后进行监听，监听连接状态
    _controller!.onConnectState.listen((bool isConnected) {
      if (mounted) {
        setState(() => _connected = isConnected);
      }
    });

取消订阅消息回调

  // 当不需要改消息监听时候，需要调用destory释放资源，取消订阅消息回调
  @override
  void dispose() {
    _controller?.destroy();
    super.dispose();
  }