使用PAHO訂閱和閱讀MQTT消息

Question

我使用paho發送和接收mqtt消息。 到目前爲止，發送消息一直沒有問題，我通過使用蚊子來接收它們。

現在我想通過使用java客戶端來讀取消息，並且我注意到關於接收消息的文檔較少。

我實現了MqttCallback接口，但我仍然無法弄清楚如何閱讀我訂閱的主題的消息。

這是我的源代碼到目前爲止，我可以閱讀使用mosquitto_sub的消息。

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

public class PahoDemo implements MqttCallback { 
    MqttClient client; 
    MqttClient subClient; 

    public PahoDemo() { 
    } 

    public static void main(String[] args) { 
     new PahoDemo().doDemo(); 
    } 

    public void doDemo() { 
     try { 
      client = new MqttClient("tcp://192.168.118.11:1883", "Sending"); 
      subClient = new MqttClient("tcp://192.168.118.11:1883", 
        "Subscribing"); 
      client.connect(); 
      subClient.connect(); 
      subClient.subscribe("foo"); 
      MqttMessage message = new MqttMessage(); 
      message.setPayload("A single message from my computer fff" 
        .getBytes()); 
      client.publish("foo", message); 
      client.disconnect(); 
      client.close(); 
      subClient.disconnect(); 
      subClient.close(); 
     } catch (MqttException e) { 
      e.printStackTrace(); 
     } 
    } 

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

    } 

    @Override 
    public void messageArrived(String topic, MqttMessage message) 
      throws Exception { 
System.out.println(message);  
    } 

    @Override 
    public void deliveryComplete(IMqttDeliveryToken token) { 
     // TODO Auto-generated method stub 

    } 

} 

Answer 1

在代理人有時間發送消息之前，您正在關閉客戶端。

此外，您不需要2個客戶端實例，您可以發送和接收只有一個。

我編輯了一下你的代碼，現在它會繼續運行並接收消息，直到你殺死它。

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

public class PahoDemo implements MqttCallback { 

MqttClient client; 

public PahoDemo() { 
} 

public static void main(String[] args) { 
    new PahoDemo().doDemo(); 
} 

public void doDemo() { 
    try { 
     client = new MqttClient("tcp://192.168.118.11:1883", "Sending"); 
     client.connect(); 
     client.setCallback(this); 
     client.subscribe("foo"); 
     MqttMessage message = new MqttMessage(); 
     message.setPayload("A single message from my computer fff" 
       .getBytes()); 
     client.publish("foo", message); 
    } catch (MqttException e) { 
     e.printStackTrace(); 
    } 
} 

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

} 

@Override 
public void messageArrived(String topic, MqttMessage message) 
     throws Exception { 
System.out.println(message); 
} 

@Override 
public void deliveryComplete(IMqttDeliveryToken token) { 
    // TODO Auto-generated method stub 

} 

} 

編輯：添加缺少的client.setCallback(this)