关于mqtt断线重连后消息停留在broker上，没有推送到客户端

Question

最近在学习mqtt，拿到例程来实验发现离线后推送的信息不能在重连后接收到，这是为什么？我用wireshark照着手册看了一遍没毛病啊，而且断线重连后订阅的消息有些能收到，有些会停留在代理上，求解！！！

 public class MQTTPublish implements MqttCallback {  

    //public static final String HOST = "tcp://10.0.0.250:1884";
    public static final String HOST = "tcp://192.168.67.130:61613"; 

    public static final String TOPIC = "MQTTtest";  
    private static final String clientid ="publisher";   

    private static final String str = "ad钙奶";
    private static MqttClient client;  

    private MqttTopic topic;  
    private String userName = "admin";  
    private String passWord = "password";  

    private MqttMessage message;  

    public MQTTPublish() throws MqttException {  
        client = new MqttClient(HOST, clientid, new MemoryPersistence());  
        connect();  
    }  

    private void connect() {  
        MqttConnectOptions options = new MqttConnectOptions();  
        options.setCleanSession(false);  
        options.setUserName(userName);  
        options.setPassword(passWord.toCharArray());  
        options.setConnectionTimeout(10);   
        options.setKeepAliveInterval(20);  
        try {  
               client.setCallback(this);  

               client.connect(options);  
               topic = client.getTopic(TOPIC);  
        } catch (Exception e) {  
               e.printStackTrace();  
        }  
    }  

    public void publish(MqttMessage message) throws MqttPersistenceException, MqttException{  
        MqttDeliveryToken token = topic.publish(message);  
        token.waitForCompletion();  
        System.out.println("Token is complete:" + token.isComplete());  
    }  

    public static void main(String[] args) throws MqttException {  
        MQTTPublish mqttpub =  new MQTTPublish();  
        mqttpub.message = new MqttMessage();  
        mqttpub.message.setQos(2);  
        mqttpub.message.setRetained(true);  
        mqttpub.message.setPayload(str.getBytes());  
        mqttpub.publish(mqttpub.message);  
        System.out.println("Ratained state:" + mqttpub.message.isRetained());  

        client.disconnect();
        System.out.println("Disconnected");
        System.exit(0);
    }

    @Override
    public void connectionLost(Throwable arg0) {
        // TODO Auto-generated method stub

    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        try {
            System.out.println("deliveryComplete---------"+ token.isComplete());  
        } catch (Exception e) {
            e.printStackTrace();
        } 
    }

    @Override
    public void messageArrived(String arg0, MqttMessage arg1) throws Exception {
        // TODO Auto-generated method stub

    }  


} 

 public class MQTTSubscribe implements MqttCallback {  

    //public static final String HOST = "tcp://10.0.0.250:1884";
    public static final String HOST = "tcp://192.168.67.130:61613";  

    public static final String TOPIC = "MQTTtest";  
    private static final String clientid = "subscriber";  
    private MqttClient client;  
    private MqttConnectOptions options;  
    private String userName = "admin";  
    private String passWord = "password";  
    private ScheduledExecutorService scheduler;  


    public void startReconnect() {  
        scheduler = Executors.newSingleThreadScheduledExecutor();  
        scheduler.scheduleAtFixedRate(new Runnable() {  
            public void run() {  
                if (!client.isConnected()) {  
                    try {  
                        client.connect(options);  
                    } catch (MqttSecurityException e) {  
                        e.printStackTrace();  
                    } catch (MqttException e) {  
                        e.printStackTrace();  
                    }  
                }  
            }  
        }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);  
    }   

    private void start() {  
        try {    
            client = new MqttClient(HOST, clientid, new MemoryPersistence());  
            options = new MqttConnectOptions();  
            options.setCleanSession(false);  
            options.setUserName(userName);  
            options.setPassword(passWord.toCharArray());  
            options.setConnectionTimeout(10);   
            options.setKeepAliveInterval(20);  
            client.setCallback(this); 
//            MqttTopic topic = client.getTopic(TOPIC);  
//            options.setWill(topic, "close".getBytes(), 2, true);  

            client.connect(options);  
            int[] Qos  = {2};  
            String[] topic1 = {TOPIC};  
            client.subscribe(topic1, Qos);  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  


    public static void main(String[] args) throws MqttException {     
        MQTTSubscribe client = new MQTTSubscribe();  
        client.start();  
    }

    public void connectionLost(Throwable cause) {  
        System.out.println("Connection lost, reconnect please!");  
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
            try {
                System.out.println("deliveryComplete---------"+ token.isComplete());  
            } catch (Exception e) {
                e.printStackTrace();
            } 
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
          System.out.println("Message arrived on topic:"+topic);  
          System.out.println("Message arrived on QoS:"+message.getQos());  
          System.out.println("Message arrived on content:"+new String(message.getPayload()));   
    }  

}  

Answer 1

对于离线消息为什么不能接收到，我想说的是 你很有可能在再次连接的时候又重新订阅了该topic。这样是接收不到离线消息的。
离线消息是订阅了某个topic的client 在断开连接以后，在接下来又重新连接的时候，任然可以收到在它断开连接的这段时间内，该topic上
的消息。这就需要将client的 setCleanSession 设置为false，这样服务器才能保留它的session，再次建立连接的时候，它就会继续使用这个session了。
注意：clientId 是不能更改的。