使用PAHO订阅并阅读MQTT消息 [英] Subscribe and Read MQTT Message Using PAHO

问题描述

我正在使用paho发送和接收mqtt消息。
到目前为止，发送消息一直没有问题，我通过使用mosquitto接收它们。

现在我想通过使用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实现MqttCallback {
 MqttClient客户端; 
 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，
订阅）; 
 client.connect（）; 
 subClient.connect（）; 
 subClient.subscribe（foo）; 
 MqttMessage message = new MqttMessage（）; 
 message.setPayload（从我的电脑一个消息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自动生成的方法存根
 
 $ 
 
 @Override 
 public void messageArrived（String topic，MqttMessage message）
 throws异常{
 System.out.println（message）; 
} 
 
 @Override 
 public void deliveryComplete（IMqttDeliveryToken token）{
 // TODO自动生成的方法存根
 
} 
 
} 
  

解决方案

客户端在代理商有时间发回消息之前。

此外，您不需要客户端的2个实例，只需一个即可发送和接收。 / p>

我已经编辑了一些代码，现在将继续运行并接收消息，直到您杀死它。

  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实现MqttCallback {
 
 MqttClient客户端; 
 
 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（从我的电脑一个消息fff
 .getBytes（））; 
 client.publish（foo，message）; 
} catch（MqttException e）{
 e.printStackTrace（）; 
} 
} 
 
 @Override 
 public void connectionLost（Throwable cause）{
 // TODO自动生成的方法存根
 
 $ 
 
 @Override 
 public void messageArrived（String topic，MqttMessage message）
 throws异常{
 System.out.println（message）; 
} 
 
 @Override 
 public void deliveryComplete（IMqttDeliveryToken token）{
 // TODO自动生成的方法存根
 
} 
 
} 
  

编辑：添加缺少的 client.setCallback （这）

I'm using paho to send and receive mqtt messages. So far it has been no problem to send the messages, I'm receiving them by using mosquitto.

Now I want to read the messages by using a java client and I noticed that there has been less documentation about receiving the messages.

I implemented the MqttCallback interface but I still couldn't figure out how to read a message of a topic I've subscribed to.

This is my sourcecode so far, I can read the messages using 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

    }

}

解决方案

You are closing the client down before the broker has time to send the message back.

Also you don't need 2 instance of the client, you can send and receive with just one.

I've edited your code a little, it now will continue to run and receive messages until you kill it.

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

}

}

EDIT: added the missing client.setCallback(this)

这篇关于使用PAHO订阅并阅读MQTT消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！