Java 智慧充电 MQTT 客户端完整可运行代码

weixin_48309090

466人浏览 · 2026-05-19 09:43:05

weixin_48309090 · 2026-05-19 09:43:05 发布

一、第一步：引入 Maven 依赖

二、MQTT 配置参数（配置文件 application.yml）

三、MQTT 配置类 + 客户端初始化

四、核心：MQTT 消息回调处理类（接收设备每秒数据、心跳、离线）

五、实体类：充电实时数据、心跳实体

六、工具类：后端下发指令（启动 / 停止充电）

七、面试可直接背的代码亮点

基于 Eclipse Paho MQTT，适配充电桩设备上报、心跳、下行指令，可直接放进 device-service 设备微服务使用，带完整注释、重连、断线监听、消息回调、幂等处理。

一、第一步：引入 Maven 依赖

<!-- MQTT Java 客户端 -->
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

<!-- 序列化、工具类 -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.83</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>

二、MQTT 配置参数（配置文件 application.yml）

mqtt:
  broker: tcp://127.0.0.1:1883  # EMQX/Mosquitto 地址端口
  username: admin
  password: public
  clientId: charge_device_service_${random.uuid}
  # 订阅通配符：监听所有设备心跳、充电数据、告警
  subscribeTopic: device/+/heartbeat,device/+/charge/data,device/+/alarm,device/+/offline
  # 下行指令前缀
  cmdTopicPrefix: device/{}/cmd/

三、MQTT 配置类 + 客户端初始化

import lombok.Data;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Data
@Configuration
public class MqttConfig {

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

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

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

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

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

    /**
     * 初始化MQTT客户端
     */
    @Bean
    public MqttClient mqttClient() throws MqttException {
        // 内存持久化，重启丢失临时消息，物联网常用
        MemoryPersistence persistence = new MemoryPersistence();

        MqttClient mqttClient = new MqttClient(broker, clientId, persistence);
        // 连接参数配置
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        // 断线自动重连
        options.setAutomaticReconnect(true);
        // 清除会话：false 断线保留订阅
        options.setCleanSession(false);
        // 心跳间隔 60s
        options.setKeepAliveInterval(60);

        // 设置回调监听
        mqttClient.setCallback(new MqttMessageCallback());

        // 连接Broker
        mqttClient.connect(options);

        // 批量订阅多个Topic
        String[] topics = subscribeTopic.split(",");
        int[] qos = new int[topics.length];
        // 全部 QoS1 至少一次送达
        for (int i = 0; i < qos.length; i++) {
            qos[i] = 1;
        }
        mqttClient.subscribe(topics, qos);

        return mqttClient;
    }
}

四、核心：MQTT 消息回调处理类（接收设备每秒数据、心跳、离线）

import com.alibaba.fastjson.JSON;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

/**
 * MQTT 回调：连接断开、消息到达、发送完成
 */
@Component
public class MqttMessageCallback implements MqttCallback {

    /**
     * 连接丢失：断线自动重连由配置 automaticReconnect 控制
     */
    @Override
    public void connectionLost(Throwable cause) {
        System.out.println("MQTT 连接断开，原因：" + cause.getMessage());
    }

    /**
     * 设备上报消息入口：每秒充电数据、心跳、告警、离线都走这里
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // 1. 获取报文内容
        String payload = new String(message.getPayload());
        System.out.println("收到MQTT消息，主题：" + topic + " 报文：" + payload);

        // 2. 按不同Topic分发处理业务
        if (topic.contains("heartbeat")) {
            // 处理 设备心跳
            handleHeartbeat(payload);
        } else if (topic.contains("charge/data")) {
            // 处理 每秒充电实时数据
            handleChargeRealData(payload);
        } else if (topic.contains("alarm")) {
            // 处理 设备故障告警
            handleDeviceAlarm(payload);
        } else if (topic.contains("offline")) {
            // 处理 设备离线遗嘱消息
            handleDeviceOffline(topic, payload);
        }
    }

    /**
     * 消息发送完成回调
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

    }

    // ====================== 业务处理方法 ======================

    /**
     * 处理每秒充电数据
     */
    private void handleChargeRealData(String payload) {
        // 1. 解析JSON为充电数据实体
        ChargeRealData data = JSON.parseObject(payload, ChargeRealData.class);

        // 2. 存入Redis：更新实时状态、实时电量电压电流
        // redisTemplate.opsForValue().set("device:real:" + data.getDeviceNo(), data);

        // 3. 存入MongoDB：时序历史数据，用于充电曲线
        // mongoTemplate.insert(data);

        // 4. 推送RabbitMQ 给订单服务、告警服务
        // rabbitTemplate.convertAndSend("charge_data_exchange", "charge.data.routing", data);

        // 5. 异常判断：过压、过温、欠压，自动下发停机指令
        if (data.getTemperature() > 60) {
            // mqtt发送停机指令
            MqttCommandUtil.sendStopCommand(data.getDeviceNo());
        }
    }

    /**
     * 处理设备心跳
     */
    private void handleHeartbeat(String payload) {
        DeviceHeartBeat beat = JSON.parseObject(payload, DeviceHeartBeat.class);
        // 更新Redis 最后心跳时间戳
        // redisTemplate.opsForValue().set("device:heart:" + beat.getDeviceNo(), System.currentTimeMillis());
    }

    /**
     * 设备告警
     */
    private void handleDeviceAlarm(String payload) {
        // 解析故障码、生成运维工单、推送短信通知
    }

    /**
     * 设备离线
     */
    private void handleDeviceOffline(String topic, String payload) {
        // 标记设备离线、更新数据库、生成告警工单
    }
}

五、实体类：充电实时数据、心跳实体

import lombok.Data;
import java.io.Serializable;

/**
 * 充电桩每秒上报充电数据实体
 */
@Data
public class ChargeRealData implements Serializable {
    // 设备编号
    private String deviceNo;
    // 枪号
    private Integer socketNo;
    // 电压
    private Double voltage;
    // 电流
    private Double current;
    // 实时功率
    private Double power;
    // 已充电度数
    private Double capacity;
    // 温度
    private Integer temperature;
    // 状态 1空闲 2充电中 3故障
    private Integer status;
    // 时间戳
    private Long timestamp;
}

/**
 * 设备心跳包实体
 */
@Data
class DeviceHeartBeat {
    private String deviceNo;
    // 信号强度
    private Integer signal;
    // 固件版本
    private String version;
    private Long timestamp;
}

六、工具类：后端下发指令（启动 / 停止充电）

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.Component;

@Component
public class MqttCommandUtil {

    @Autowired
    private MqttClient mqttClient;

    private static final String CMD_START_TOPIC = "device/{}/cmd/start";
    private static final String CMD_STOP_TOPIC = "device/{}/cmd/stop";

    /**
     * 下发启动充电指令
     */
    public static void sendStartCommand(String deviceNo) throws MqttException {
        String topic = String.format(CMD_START_TOPIC, deviceNo);
        MqttMessage message = new MqttMessage();
        message.setPayload("start".getBytes());
        // QoS1 确保送达
        message.setQos(1);
        mqttClient.publish(topic, message);
    }

    /**
     * 下发停止充电指令
     */
    public static void sendStopCommand(String deviceNo) throws MqttException {
        String topic = String.format(CMD_STOP_TOPIC, deviceNo);
        MqttMessage message = new MqttMessage();
        message.setPayload("stop".getBytes());
        message.setQos(1);
        mqttClient.publish(topic, message);
    }
}