Cannot receive already published messages to subsc

问题:

I'm using paho to send and receive mqtt messages. So far it has been no problem to send the messages. I have problems with receiving them.My code is:

     package BenchMQTT;

     import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
     import org.eclipse.paho.client.mqttv3.IMqttToken;
     import org.eclipse.paho.client.mqttv3.MqttCallback;
     import org.eclipse.paho.client.mqttv3.MqttException;
     import org.eclipse.paho.client.mqttv3.MqttMessage;
     import org.eclipse.paho.client.mqttv3.MqttClient;

     public class Test_A_2 implements MqttCallback {

     MqttClient clientR;
     MqttClient clientS;

     public Test_A_2() {
     }

     public static void main(String[] args) throws InterruptedException {
         long startTime = System.currentTimeMillis();
         new Test_A_2().doDemo();
         long endTime = System.currentTimeMillis();
     }

    public void doDemo() throws InterruptedException {
    try {   
    clientS = new MqttClient("tcp://mybroker:1883", "Sender");
    clientR = new MqttClient("tcp://mybroker:1883", "Reiever");
    clientR.connect();
    clientS.connect();
    MqttMessage message = new MqttMessage();

    String messagePayload = "qwertyuiopasdfghjklzxcvbnmqwertyuiopasdfghjk"
            + "lzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasdfghj"
            + "klzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasdfgh"
            + "jklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasdfg"
            + "hjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasd"
            + "fghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopas"
            + "dfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopa"
            + "sdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiop"
            + "asdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuio"
            + "pasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyui"
            + "opasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyu"
            + "iopasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwerty"
            + "uiopasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwert"
            + "nmqwertyuiop";

    clientR.subscribe("BenchMQTT");   
    clientR.setCallback(this);

    for(int i=0;i<10;i++)
    {
    message.setPayload((messagePayload)
            .getBytes());
    System.out.println(i);
    clientS.publish("BenchMQTT", message);
    }
    clientR.disconnect();   
    clientS.disconnect();
    clientR.close();   
    clientS.close();

   } catch (MqttException e)
    {
     System.out.println("ERROR");
    }
 }

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

     }

     @Override
     public void messageArrived(String topic, MqttMessage message)
     {
         System.out.println("Received: " + message.toString());
     }

     @Override
     public void deliveryComplete(IMqttDeliveryToken token) {

     }

     }

This send and receives messages.

OUTPUT:

0
Received: 0
1
Received: 1
2
Received: 2
3
Received: 3
4
Received: 4
5
Received: 5
6
Received: 6
7
Received: 7
8
Received: 8
9
Received: 9

I would like to send messages, and after that receive them. Any help? Expected OUTPUT:

0
1
2
3
4
5
6
7
8
9
Received: 0
Received: 1
Received: 2
Received: 3
Received: 4
Received: 5
Received: 6
Received: 7
Received: 8
Received: 9

回答1:

That's not how MQTT (or any pub/sub messaging) works, if the receiver is connected to the server then messages will be delivered as they are sent.

The exception to this is if the receiver connects and subscribes to topic with a QOS greater than 0 then disconnects and reconnects later (with out the clean session flag set) then the missed messages that have been published with a QOS greater than 0 will be delivered at the point of reconnect.

The other possibility is if the messages have been published with the retained flag set true, but then only the last message published to a topic will be delivered at the point the receiving client subscribes.

回答2:

The following code does what you want, but it is forcing MQTT to behave in a way it is not supposed to. Message queuing on the is only intended to ensure all messages are delivered to a client even if it is disconnected for a period, messages will always be delivered at the earliest possible opportunity.

 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.IMqttToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.eclipse.paho.client.mqttv3.MqttClient;

 public class Test_A_2 implements MqttCallback {

 MqttClient clientR;
 MqttClient clientS;

 public Test_A_2() {
 }

 public static void main(String[] args) throws InterruptedException {
     long startTime = System.currentTimeMillis();
     new Test_A_2().doDemo();
     long endTime = System.currentTimeMillis();
 }

public void doDemo() throws InterruptedException {
try {   

    MqttConnectOptions options = new MqttConnectOptions();
    options.setCleanSession(false);

clientS = new MqttClient("tcp://localhost:1883", "Sender");
clientR = new MqttClient("tcp://localhost:1883", "Reiever");
clientR.connect(options);
clientS.connect();
clientR.setCallback(this);
clientR.subscribe("BenchMQTT",2);
MqttMessage message = new MqttMessage();

String messagePayload = "qwertyuiopasdfghjklzxcvbnmqwertyuiopasdfghjk"
        + "lzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasdfghj"
        + "klzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasdfgh"
        + "jklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasdfg"
        + "hjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopasd"
        + "fghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopas"
        + "dfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiopa"
        + "sdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuiop"
        + "asdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyuio"
        + "pasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyui"
        + "opasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwertyu"
        + "iopasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwerty"
        + "uiopasdfghjklzxcvbnmqwertyuiopasdfghjklzxcvbnmqwert"
        + "nmqwertyuiop";

clientR.disconnect();

for(int i=0;i<10;i++)
{
message.setPayload((messagePayload)
        .getBytes());
System.out.println(i);
message.setQos(2);
clientS.publish("BenchMQTT", message);
}



clientR.connect(options);
clientR.setCallback(this);
clientR.subscribe("BenchMQTT",2);

clientR.disconnect();   
clientS.disconnect();
clientR.close();   
clientS.close();

} catch (MqttException e)
{
 System.out.println("ERROR");
 e.printStackTrace();
}
}

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

 }

 @Override
 public void messageArrived(String topic, MqttMessage message)
 {
     System.out.println("Received: " + message.toString());
 }

 @Override
 public void deliveryComplete(IMqttDeliveryToken token) {

 }

 }