4

我正在使用 paho 发送和接收 mqtt 消息。到目前为止，发送消息没有问题。我收到它们时遇到问题。我的代码是：

     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) {

     }

     }

这发送和接收消息。

输出：

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

我想发送消息，然后接收它们。有什么帮助吗？预期输出：

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
4

2 回答 2

4

以下代码可以满足您的需求，但它会强制 MQTT 以不应该的方式运行。消息队列只是为了确保所有消息都被传递给客户端，即使它在一段时间内断开连接，消息也将始终尽可能早地传递。

 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) {

 }

 }
于 2014-04-30T21:20:32.940 回答
4

这不是 MQTT（或任何 pub/sub 消息传递）的工作方式，如果接收器连接到服务器，则消息将在发送时传递。

例外情况是，如果接收者连接并订阅了 QOS 大于 0 的主题，然后断开连接并稍后重新连接（没有设置干净会话标志），则已发布的 QOS 大于 0 的丢失消息将被传递在重新连接点。

另一种可能性是，如果消息已在将保留标志设置为 true 的情况下发布，但只有发布到主题的最后一条消息将在接收客户端订阅时被传递。

于 2014-04-30T14:46:57.113 回答