客户如何知道它是否已经订阅了MQTT主题？

Question

我正在订阅一个MQTT主题（在我的例子中它是应用程序唯一的用户ID）。我正在使用AWS IOT核心服务进行订阅。每当主屏幕打开并且我从awsConnectClient接收回调时，我都会拨打订阅号码。现在发生了什么，如果应用程序开放三次它订阅相同的主题3次。现在任何消息发布到该主题。它收到应用程序3次。

现在我想做什么，我想知道如果这个userId已经从这个设备订阅，我将不会再次从同一设备拨打电话。

一种方法可能是如果我在我的应用程序中保存，我已经订阅了这个主题，并且不再拨打订阅电话。但是我怀疑这种方法对于所有场景是否都是正确的。如果有任何问题，我们可以从服务器端驱动这个逻辑，如果有任何问题，我可以告诉我这已经订阅了。

fun connectClick() {
    Log.d(TAG, "clientId = $clientId")

    try {
        mqttManager.connect(clientKeyStore) { status, throwable ->
            Log.d(TAG, "Status = " + status.toString())
            var formattedStatus = String.format(getString(R.string.status_msg),status.toString())

            if (status == AWSIotMqttClientStatusCallback.AWSIotMqttClientStatus.Connected) {
                Log.i(TAG, " subscribed to - " + VoiceXPreference(this).rosterName)
                unsubscribe()
                subscribeClick(VoiceXPreference(this).rosterName)
            }
            runOnUiThread {
                tv_iot_status.text = formattedStatus
                if (throwable != null) {
                    Log.e(TAG, "Connection error.", throwable)
                }
            }
        }
    } catch (e: Exception) {
        Log.e(TAG, "Connection error.", e)
    }

}

以上是我的订阅代码。虽然我在订阅之前总是取消订阅，但这对我不起作用。

以下是我的initClient调用，它发出连接请求。我添加了if检查mqttManager是否已经初始化首先断开连接然后发出连接请求。虽然我已将initRequest放入onCreate（）回调应用程序屏幕，该应用程序屏幕仅在应用程序打开时调用一次。我已经检查了它只被调用一次的日志。

AWSMobileClient.getInstance().initialize(this, object : Callback<UserStateDetails> {
            override fun onResult(result: UserStateDetails) {
                Log.i(TAG,"connect request called");
                if(mqttManager != null){
                    mqttManager?.disconnect()
                }
                initIoTClient()
            }

            override fun onError(e: Exception) {
                Log.e(TAG, "onError: ", e)
            }
        })

以下是我的订阅代码片段，订阅了唯一的userId

fun subscribeClick(topic: String) {

    Log.d(TAG, "topic = $topic")

    try {
        mqttManager?.subscribeToTopic(topic, AWSIotMqttQos.QOS0,
            { topic, data ->
                runOnUiThread {
                    try {
                        val message = String(data, Charsets.UTF_8)
                        Log.d(TAG, "Message arrived:")
                        Log.d(TAG, "   Topic: $topic")
                        Log.d(TAG, " Message: $message")

                        val gson = Gson()
                        val notificationModel = gson.fromJson(message, NotificationModel::class.java)
                        var orderServiceMapperResponseModel = OrderServiceMapperResponseModel()
                        orderServiceMapperResponseModel.seatId = notificationModel.seatId
                        orderServiceMapperResponseModel.serviceName = notificationModel.service
                        orderServiceMapperResponseModel.id = notificationModel.id
                        orderServiceMapperResponseModel.createdDate = notificationModel.createdDate
                        serviceList.add(orderServiceMapperResponseModel)
                        if (isPictureInPictureMode) {
                            if (isShownNotification) {
                                updateNotificationCount()
                            } else {
                                updatePIPWindowContent()
                            }
                        } else {
                            updateAdapterDataSource()
                        }

                    } catch (e: UnsupportedEncodingException) {
                        Log.e(TAG, "Message encoding error.", e)
                    }
                }
            })
    } catch (e: Exception) {
        Log.e(TAG, "Subscription error.", e)
    }
}

我也总是在我的应用程序屏幕的onDestroy（）内部发出disconnect（）请求

mqttManager?.disconnect()

但我仍然得到3条订阅消息而不是1条消息。

Answer 1

您收到3条重复邮件，不是因为您订阅了3次，而是因为您创建了3个单独的连接。

MQTT specification明确指出

如果服务器收到包含与现有Subscription主题过滤器相同的主题过滤器的SUBSCRIBE数据包，那么它必须用新的订阅完全替换现有的Subscription。

意味着每个连接的重复订阅永远不会发生，除非服务器具有损坏的实现。

您的代码看起来就像在调用代码块时创建新连接时从不发送断开连接请求。

您应该保留单个MQTT会话，或确保在关闭应用程序时关闭连接。