1.pom

<!-- mqtt -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
</dependency>

2. config

package com.shitong.web.core.config;

import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@Configuration
public class MqttConfig {

    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.url}")
    private String host;

    @Value("${spring.mqtt.client.id}")
    private String clientId;

    @Value("${spring.mqtt.default.topic}")
    private String topic;

    // 配置MQTT客户端工厂
    @Bean
    public MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{host});
        mqttConnectOptions.setKeepAliveInterval(20);
        mqttConnectOptions.setAutomaticReconnect(true); // 开启自动重连
        return mqttConnectOptions;
    }

    // 配置MQTT客户端
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }

    // 订阅通道
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    // 发布通道
    @Bean
    public MessageChannel mqttOutChannel() {
        return new DirectChannel();
    }

    // 订阅适配器
    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttClientFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        clientId + "-inbound",
                        mqttClientFactory,
                        topic, "$SYS/brokers/+/clients/#"); // 同时订阅业务主题和系统主题
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    // 发布适配器
    @Bean
    @ServiceActivator(inputChannel = "mqttOutChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(
                clientId + "-outbound", mqttClientFactory());
        handler.setAsync(true); // 异步发送
        handler.setDefaultQos(1); // 设置QoS级别
        return handler;
    }
}

3.监听器

package com.shitong.web.core.config;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.shitong.energy.service.EnergyDataService;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageHandler;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.*;

/**
 * MQTT消息监听器
 * @author shitong
 */
@Component
public class MqttMessageListener {

    @Resource
    private EnergyDataService service;

  Logger log = LoggerFactory.getLogger(MqttMessageListener.class);

    // 处理入站消息
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            String topic = Objects.requireNonNull(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)).toString();
            String payload = message.getPayload().toString();
          
            log.info("[MQTT] 收到消息: topic={}, payload={}", topic, payload);
            try {
                if (topic.startsWith("$SYS/brokers/")) {
                    // 处理客户端上下线事件
                    handleClientLifecycleEvent(topic, payload);

                } else if (topic.startsWith("data/device_id")) {
                    String ip = getActiveNetworkIP();
                    System.out.println("数据上报");
                }
            } catch (Exception e) {
                log.error("[MQTT] 消息处理失败: {}", e.getMessage());
            }
        };
    }

    /**
     * 处理客户端上下线事件
     */
    private void handleClientLifecycleEvent(String topic, String payload) {
        // 解析主题格式：$SYS/brokers/{node}/clients/{clientId}/connected 或 .../disconnected
        String[] parts = topic.split("/");
        if (parts.length < 6) return;
        String clientId = parts[4];  // 客户端ID
        String action = parts[5];   // connected/disconnected
        if ("connected".equals(action)) {
            log.info("[MQTT] 客户端上线: clientId={}", clientId);
            // 上线逻辑：更新状态、发送通知等
        } else if ("disconnected".equals(action)) {
            log.info("[MQTT] 客户端下线: clientId={}", clientId);
            // 下线逻辑：更新状态、发送告警等
        }
    }

    private static String getActiveNetworkIP() throws Exception {
        Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
        while (interfaces.hasMoreElements()) {
            NetworkInterface networkInterface = interfaces.nextElement();
            if (networkInterface.isLoopback() || !networkInterface.isUp()) {
                continue;
            }
            Enumeration<InetAddress> addresses = networkInterface.getInetAddresses();
            while (addresses.hasMoreElements()) {
                InetAddress address = addresses.nextElement();
                if (!address.isLoopbackAddress() &&
                        !address.isLinkLocalAddress() &&
                        address.isSiteLocalAddress()) {
                    return address.getHostAddress();
                }
            }
        }
        throw new Exception("未找到有效网络接口");
    }
}

4.yml地址

spring:
  mqtt:
    # MQTT服务器URL
    url: tcp://xxx.xxx.xxx.xxx:1883
    # MQTT用户名
    username: admin
    # MQTT密码
    password: public
    # MQTT主题
    default:
      topic: data/#
    client:
      id: mqtt_spring