在Spring Boot项目中集成MQTT并实现发布/订阅功能，你可以使用Eclipse Paho MQTT客户端库。以下是一个简单的步骤指南和示例代码，帮助你开始在Spring Boot中集成MQTT。

步骤指南

1. 添加依赖：在你的pom.xml文件中添加Eclipse Paho MQTT客户端的依赖。

【xml】

org.eclipse.paho
org.eclipse.paho.client.mqttv3
1.2.5

注意：版本号可能会更新，请检查Maven中央仓库以获取最新版本。

2. 配置MQTT客户端：创建一个配置类来设置MQTT客户端的连接参数，如服务器URI、客户端ID、用户名、密码等（如果需要）。

3. 实现发布和订阅逻辑：使用MqttClient类来连接到MQTT服务器，并使用publish方法发布消息，使用subscribe方法订阅主题。

4. 创建Spring Bean：将MQTT客户端配置为Spring Bean，以便在应用程序的其他部分注入和使用。

5. 处理消息回调：实现MqttCallback接口来处理接收到的消息。

示例代码

以下是一个简单的Spring Boot应用程序，它集成了MQTT并实现了一个发布者和订阅者。

MQTT配置类

【java】
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MqttConfig {

private static final String BROKER_URL = "tcp://mqtt.eclipse.org:1883";
private static final String CLIENT_ID = "springboot-mqtt-client";

@Bean
public MqttClient mqttClient() throws MqttException {
    MqttClient client = new MqttClient(BROKER_URL, CLIENT_ID, new MemoryPersistence());
    MqttConnectOptions options = new MqttConnectOptions();
    options.setCleanSession(true);
    client.connect(options);
    return client;
}

}

消息发布者

【java】
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MqttPublisher {

@Autowired
private MqttClient mqttClient;

public void publish(String topic, String message) throws MqttException {
    MqttMessage mqttMessage = new MqttMessage(message.getBytes());
    mqttMessage.setQos(2); // 设置QoS等级
    mqttClient.publish(topic, mqttMessage);
}

}

消息订阅者

【java】
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;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MqttSubscriber implements MqttCallback {

@Autowired
private MqttClient mqttClient;

public void subscribe(String topic) throws MqttException {
    mqttClient.setCallback(this);
    mqttClient.subscribe(topic);
}

@Override
public void connectionLost(Throwable cause) {
    // 处理连接丢失
}

@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
    // 处理接收到的消息
    System.out.println("Message arrived. Topic: " + topic + " Message: " + new String(message.getPayload()));
}

@Override
public void deliveryComplete(IMqttDeliveryToken token) {
    // 处理消息发送完成
}

}

使用示例

【java】
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

@Component
public class MqttRunner implements CommandLineRunner {

@Autowired
private MqttPublisher mqttPublisher;

@Autowired
private MqttSubscriber mqttSubscriber;

@Override
public void run(String... args) throws Exception {
    try {
        mqttSubscriber.subscribe("test/topic");
        mqttPublisher.publish("test/topic", "Hello MQTT!");
    } catch (MqttException e) {
        e.printStackTrace();
    }
}

}

在这个示例中，MqttRunner类实现了CommandLineRunner接口，以便在Spring Boot应用程序启动时运行。它订阅了"test/topic"主题，并发布了一条消息到该主题。

请确保你的Spring Boot应用程序已经正确配置，并且你有权访问MQTT服务器（在这个例子中是mqtt.eclipse.org）。如果你使用自己的MQTT服务器，请相应地更改BROKER_URL。