Spring Boot 整合 MQTT 服务器实现消息发送与订阅

引入 MQTT 依赖

• 使用 org.eclipse.paho.client.mqttv3 作为 MQTT 客户端库，版本建议为 1.2.4。
• 在项目的 pom.xml 中添加以下依赖：

        
    
         
          org.eclipse.paho
         
    
         
          org.eclipse.paho.client.mqttv3
         
    
         
          1.2.4
         
        

引入 Guava 工具库

• 若需要线程池和并发处理功能，可引入 com.google.guava，版本建议为 32.1.2-jre。
• 在 pom.xml 中添加：

        
    
         
          com.google.guava
         
    
         
          guava
         
    
         
          32.1.2-jre
         
    
         
          provided
         
        

引入 JSON 解析工具

• 使用 fastjson 进行 JSON 格式数据解析，版本建议为 1.2.75。
• 添加以下依赖：

        
    
         
          com.alibaba
         
    
         
          fastjson
         
    
         
          1.2.75
         
        

引入 Lombok 工具

• 使用 lombok 进行代码简化和格式化，添加以下依赖：

        
    
         
          org.projectlombok
         
    
         
          lombok
         
    
         
          true
         
        

创建 application.yml 文件

• 在项目根目录下创建 application.yml，添加 MQTT Broker 配置：

  badao-mqtt:
    host: 127.0.0.1
    port: 1883
    user: admin
    password: 123456

创建配置类

• 创建 com.badao.demo.config.MqttConfig 类，获取上述配置：

  @Component("MqttConfig")
  @ConfigurationProperties(prefix = "badao-mqtt")
  public class MqttConfig {
    private static String host;
    private static String port;
    private static String user;
    private static String password;
    public String getHost() { return host; }
    public void setHost(String host) { MqttConfig.host = host; }
    public String getPort() { return port; }
    public void setPort(String port) { MqttConfig.port = port; }
    public String getUser() { return user; }
    public void setUser(String user) { MqttConfig.user = user; }
    public String getPassword() { return password; }
    public void setPassword(String password) { MqttConfig.password = password; }
  }

消息发送

• 使用 MqttClient 发送消息至指定主题。
• 示例代码：

  @Service
  public class MqttSender {
    @Autowired
    private MqttConfig mqttConfig;
    public void sendWarningMessage(String topic, String message) {
      final String host = mqttConfig.getHost();
      final String port = mqttConfig.getPort();
      final String username = mqttConfig.getUser();
      final String password = mqttConfig.getPassword();
      try {
        MqttClient client = new MqttClient(host, port, username, password);
        client.connect();
        MqttMessage messageObj = new MqttMessage();
        messageObj.setTopic(topic);
        messageObj.setPayload(message);
        new Thread(() -> {
          try {
            client.publish(messageObj);
          } catch (MqttException e) {
            log.error(" MQTT 发送失败：" + e.getMessage());
          }
        }).start();
        client.disconnect();
      } catch (MqttException e) {
        log.error(" MQTT 连接错误：" + e.getMessage());
      }
    }
  }

消息订阅

• 创建订阅器类，监听指定主题的消息。
• 示例代码：

  @Service
  public class MqttSubscriber {
    @Autowired
    private MqttConfig mqttConfig;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private MySQLDAO mysqlDAO;
    public void subscribe(String topic) {
      final String host = mqttConfig.getHost();
      final String port = mqttConfig.getPort();
      final String username = mqttConfig.getUser();
      final String password = mqttConfig.getPassword();
      try {
        MqttClient client = new MqttClient(host, port, username, password);
        client.connect();
        client.subscribe(topic, new MqttMessageCallback() {
          @Override
          public void onSuccess(MqttMessage message) {
            log.info("接收到消息：" + message.toString());
            // 存储消息至数据库或缓存系统
            this.redisTemplate.convertAndStore(message);
            this.mysqlDAO.saveMessage(message);
          }
          @Override
          public void onFailure(int code, String reason) {
            log.error("订阅失败：" + reason);
          }
          @Override
          public void onDisconnected() {
            log.info("与 MQTT Broker 连接丢失");
          }
        });
      } catch (MqttException e) {
        log.error(" MQTT 订阅错误：" + e.getMessage());
      } finally {
        client.disconnect();
      }
    }
  }