长连接通道

更新时间：2025-11-24 09:46:09

长连接通道SDK，提供IoT业务协议封装的云端数据下行能力。为App提供订阅消息的能力和支持请求响应模型。

SDK封装了上行RPC 订阅、取消订阅等接口 。

目前SDK仅支持设备维度订阅，暂不支持自定义topic

package com.polaris.iot.appsdk.mqtt.client

/**
 * MqttClient管理器
 */
class IoTMqttClientMgr private constructor(private val application: Application) {
    companion object {
        fun getInstance(application: Application): IoTMqttClientMgr
    }
    /**
     * 获取设备MQTT通道
     * @param deviceInfo 基本信息
     * @return 获取失败时返回 null
     */
    fun getMqttClient(deviceInfo: SampleDeviceInfo): IoTMqttClient?
}

/**
 * MQTT客户端
 */
class IoTMqttClient private constructor(private val deviceInfo: SampleDeviceInfo) {
    /**
     * 客户端是否已连接
     */
    val isConnected: Boolean
	/**
     * 设置订阅消息监听
     * @param listener 消息监听回调
     */
    fun addOnMessageListener(listener: OnMqttMessageListener)
	
    /**
     * 移除订阅消息监听
     * @param listener 消息监听回调
     */
    fun removeOnMessageListener(listener: OnMqttMessageListener)
    
    /**
     * 订阅设备消息
     * @param listener 结果回调
     */
    fun subscribe(listener: OnResultListener<Unit>)
    /**
     * 取消订阅
     * @param listener 结果回调
     */
    fun unsubscribe(listener: OnResultListener<Unit>)
}


/**
 * MQTT订阅消息回调
 */
interface OnMqttMessageListener {
    /**
     * 接收订阅消息
     * @param topic 接收到的主题（暂未使用）
     * @param message 接收到消息
     */
    fun onMessageArrived(topic: String, message: String)
}

添加依赖

打开项目的 build.gradle，添加mqtt依赖

dependencies {
    ...
    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.4'
}

配置AndroidManifest.xml

<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.WAKE_LOCK" />
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
<uses-permission android:name="android.permission.USE_EXACT_ALARM" />

<application>
    ...
	<service android:name="com.polaris.iot.appsdk.mqtt.service.MqttService" />
</application>

混淆

#mqtt
-keep class org.eclipse.paho.client.mqttv3.** {*;}
-keep interface org.eclipse.paho.client.mqttv3.** {*;}

获取客户端

private IoTMqttClient mMqttClient;

@Override
public void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    DeviceInfo deviceInfo = (DeviceInfo) getIntent().getSerializableExtra(EXTRA_DEVICE_INFO);
    //获取客户端
    mMqttClient = IoTMqttClientMgr.getInstance(getApplication()).getMqttClient(deviceInfo);
    if (mMqttClient == null) {
        showAlert("get client failed");
        return;
    }
 }

订阅

mMqttClient.subscribe(new OnResultListener<Unit>() {
    @Override
    public void onResult(Unit result) {
		//订阅成功
    }

    @Override
    public void onError(@NotNull IoTSDKError error) {
		//订阅失败
    }
});

取消订阅

@Override
public void onDestroy() {
    super.onDestroy();
    if (mMqttClient != null) {
        //取消订阅
        mMqttClient.unsubscribe(new OnResultListener<Unit>() {
            @Override
            public void onResult(Unit result) {
                //取消订阅成功
            }

            @Override
            public void onError(@NotNull IoTSDKError error) {
                //取消订阅失败
            }
        });
}

设置订阅消息回调

private onMqttMessageListener onMqttMessageListener = new OnMqttMessageListener() {
    @Override
    public void onMessageArrived(@NotNull String topic, @NotNull String message) {
        runOnUiThread(new Runnable() {
            @Override
            public void run() {
                mMqttMsgListAdapter.getDataList().add(0, new MqttMsgListAdapter.MqttMessage(message, System.currentTimeMillis(), false));
                mMqttMsgListAdapter.notifyDataSetChanged();
            }
        });
    }
};

mMqttClient.setOnMessageListener(onMqttMessageListener);

取消订阅消息回调

@Override
public void onDestroy() {
    super.onDestroy();
    mMqttClient.removeOnMessageListener(onMqttMessageListener);
}