MQTT消息在取消订阅后仍然存在，并在再次订阅时重复使用

Question

   private void doConnect() {
            Log.d(TAG, "doConnect()");
            IMqttToken token;
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setKeepAliveInterval(30);
            try {
            mqttClient = new MqttAsyncClient(tcp_server_URL, deviceId, new MemoryPersistence());

            token = mqttClient.connect(options);
            token.waitForCompletion(3500);
            mqttClient.setCallback(new MqttEventCallback());
            //changed single subscription of each topic to mass subscription
            // using String [] of Topics and int [] of QoS

            if (channelList != null) {
            channelListStringArray= new String[channelList.size()];
            channelListQosStringArray= new int[channelList.size()];
            for (int i = 0; i < channelList.size(); i++) {
            // creating String Array of topics and int Array of QoS
            channelListStringArray [i] = "account/" + channelList.get(i);
            channelListQosStringArray [i] = 1;
            }
            //subscribe all channels by passing all topics as String Array and QoS int array
            token = mqttClient.subscribe(channelListStringArray, channelListQosStringArray);
            token.waitForCompletion(3500);
            }

            } catch (MqttSecurityException e) {
            e.printStackTrace();
            } catch (MqttException e) {
            switch (e.getReasonCode()) {
            case MqttException.REASON_CODE_BROKER_UNAVAILABLE:
            case MqttException.REASON_CODE_CLIENT_TIMEOUT:
            case MqttException.REASON_CODE_CONNECTION_LOST:
            case MqttException.REASON_CODE_SERVER_CONNECT_ERROR:
            Log.v(TAG, "c" + e.getMessage());
            e.printStackTrace();
            break;
            case MqttException.REASON_CODE_FAILED_AUTHENTICATION:
            Intent i = new Intent("RAISEALLARM");
            i.putExtra("ALLARM", e);
            Log.e(TAG, "b" + e.getMessage());
            break;
            default:
            Log.e(TAG, "a" + e.getMessage());
            }
            }
            }

    public void onDestroy() {
            super.onDestroy();
            //disconnect from mqtt server
            IMqttToken token;
            try {
            if (channelList != null) {
            channelListStringArray = new String[channelList.size()];
            for (int i = 0; i < channelList.size(); i++) {
            //creating array of topics
            channelListStringArray[i] = "account/" + channelList.get(i);
            }
            //unsubscribe MQTTClient by passing String array of Topics, NULL, mqttActionListener to get CallBack
            //on success or failure of unsubscription
            token = mqttClient.unsubscribe(channelListStringArray, getApplicationContext(), mqttActionListener);
            token.waitForCompletion(3500);
            }
            //Disconnect MqttCient
            token = mqttClient.disconnect();
            token.waitForCompletion(3500);
            } catch (MqttException e) {
            e.printStackTrace();
            }
            unregisterReceiver(broadcastReceiver);
            }

Answer 1

这是因为你把洁净标志设为假。

如果不希望消息排队，则将其设置为true。

...
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(true);
options.setKeepAliveInterval(30);
...

Clean Session标志告诉代理存储QoS1/2订阅的任何消息，直到客户端重新连接。