<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <!-- Eclipse Paho MQTT Client -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.5</version>
    </dependency>
</dependencies>

mqtt:
  broker-url: tcp://localhost:1883
  client-id: springboot-mqtt-client
  username: mqtt_user
  password: mqtt_password
  topic: test/topic
  qos: 1

import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MqttConfig {
    @Value("${mqtt.broker-url}")
    private String brokerUrl;
    @Value("${mqtt.client-id}")
    private String clientId;
    @Value("${mqtt.username}")
    private String username;
    @Value("${mqtt.password}")
    private String password;
    @Bean
    public MqttClient mqttClient() throws MqttException {
        MqttClient client = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(20);
        // 設(shè)置回調(diào)
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("MQTT連接丟失：" + cause.getMessage());
            }
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("收到消息 - 主題：" + topic + "，內(nèi)容：" + new String(message.getPayload()));
            }
            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("消息發(fā)送完成");
            }
        });
        // 連接到MQTT Broker
        client.connect(options);
        return client;
    }
}

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MqttPublishService {
    @Autowired
    private MqttClient mqttClient;
    @Value("${mqtt.topic}")
    private String topic;
    @Value("${mqtt.qos}")
    private int qos;
    public void publish(String payload) throws Exception {
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(qos);
        mqttClient.publish(topic, message);
        System.out.println("消息已發(fā)布 - 主題：" + topic + "，內(nèi)容：" + payload);
    }
}

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MqttSubscribeService {
    @Autowired
    private MqttClient mqttClient;
    @Value("${mqtt.topic}")
    private String topic;
    @Value("${mqtt.qos}")
    private int qos;
    public void subscribe() throws Exception {
        mqttClient.subscribe(topic, qos);
        System.out.println("已訂閱主題：" + topic);
    }
}

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping("/mqtt")
public class MqttController {
    @Autowired
    private MqttPublishService mqttPublishService;
    @Autowired
    private MqttSubscribeService mqttSubscribeService;
    @PostMapping("/publish")
    public String publishMessage(@RequestParam String message) {
        try {
            mqttPublishService.publish(message);
            return "消息發(fā)布成功";
        } catch (Exception e) {
            return "消息發(fā)布失敗：" + e.getMessage();
        }
    }
    @GetMapping("/subscribe")
    public String subscribeTopic() {
        try {
            mqttSubscribeService.subscribe();
            return "訂閱成功";
        } catch (Exception e) {
            return "訂閱失?。? + e.getMessage();
        }
    }
}