如何使用Eclipse Paho在Java MQTT客户端上接收消息时发布消息

在使用Eclipse Paho客户端库来开发Java MQTT应用时，我们可以同时实现消息的发布与接收。这通常需要两个主要步骤：设置一个MQTT客户端，以及创建一个回调来处理接收到的消息并根据需要发布消息。下面是这一过程的具体步骤和代码示例。

步骤 1: 设置 MQTT 客户端

首先，我们需要建立一个MQTT客户端，连接到MQTT服务器。我们可以使用 MqttClient类来做到这一点。

java
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttPublishReceive {
    public static void main(String[] args) {
        String broker = "tcp://mqtt.example.com:1883";
        String clientId = "JavaClient";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient mqttClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            mqttClient.connect(connOpts);
            System.out.println("Connected to broker: " + broker);
          
            // 稍后设置回调
        } catch(MqttException me) {
            System.out.println("reason " + me.getReasonCode());
            System.out.println("msg " + me.getMessage());
            System.out.println("loc " + me.getLocalizedMessage());
            System.out.println("cause " + me.getCause());
            System.out.println("excep " + me);
            me.printStackTrace();
        }
    }
}

步骤 2: 设置消息回调

在客户端成功连接后，我们需要设置一个回调函数，该函数将在接收到消息时触发。在这个回调中，我们可以处理接收到的消息，并根据需要发布新的消息。

java
import org.eclipse.paho.client.mqttv3.MqttCallback;

public class MqttCallbackExample implements MqttCallback {
    MqttClient client;

    public MqttCallbackExample(MqttClient client) {
        this.client = client;
    }

    @Override
    public void connectionLost(Throwable cause) {
        System.out.println("Connection lost!");
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("Received a message: " + new String(message.getPayload()));
        // 对接收到的消息进行处理并发布新的消息
        String responseTopic = "response/topic";
        String responseContent = "Processed message: " + new String(message.getPayload());
        publishMessage(responseTopic, responseContent);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("Delivery Complete!");
    }

    private void publishMessage(String topic, String content) throws MqttException {
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(2);
        client.publish(topic, message);
        System.out.println("Message published to topic: " + topic);
    }
}

然后，在主程序中，你需要注册这个回调：

java
mqttClient.setCallback(new MqttCallbackExample(mqttClient));
mqttClient.subscribe("input/topic", 1);

示例结束

这样，你的Java MQTT客户端就可以在接收到消息的同时，根据接收到的消息内容发布新的消息。这对于需要实时响应外部信息的系统特别有用，例如物联网(IoT)应用。