从 0 到 1 实现 Java 与语音设备的 MQTT 通信：物联网语音交互全攻略

# 安装Mosquitto
sudo apt update
sudo apt install mosquitto mosquitto-clients

# 启动服务并设置开机自启
sudo systemctl start mosquitto
sudo systemctl enable mosquitto

# 验证运行状态
sudo systemctl status mosquitto  # 显示active(running)即成功

# 允许远程连接（默认仅本地）
listener 1883 0.0.0.0

# 开启匿名访问（开发环境，生产环境需关闭）
allow_anonymous true

# 日志配置
log_dest file /var/log/mosquitto/mosquitto.log
log_level info

# 下载并安装
curl
-s
 https://assets.emqx.com/scripts/install-emqx-deb.sh 
|
sudo
bash
sudo
apt
install
 emqx
# 启动服务
sudo
 emqx start  
# 输出"EMQX 5.1.0 is started successfully!"即成功
# 访问控制台（默认账号admin/public）
# 地址：http://localhost:18083

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

<!-- Spring Boot基础依赖（非必须，根据项目类型选择） -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>

<!-- Lombok简化代码 -->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>

<!-- 单元测试依赖 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

package com.example.mqtt.config;

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

/**
 * MQTT连接参数配置类
 * 从配置文件读取Broker地址、端口、客户端ID等参数
 */
@Data
@Component
@ConfigurationProperties(prefix = "mqtt")
public class MqttProperties {
    /**
     * MQTT Broker地址（如tcp://localhost:1883）
     */
    private String brokerUrl;

    /**
     * 客户端ID（需唯一，建议包含设备标识）
     */
    private String clientId;

    /**
     * 用户名（Broker开启认证时必填）
     */
    private String username;

    /**
     * 密码（Broker开启认证时必填）
     */
    private String password;

    /**
     * 连接超时时间（秒）
     */
    private int connectionTimeout = 30;

    /**
     * 会话保持时间（秒）
     */
    private int keepAliveInterval = 60;

    /**
     * 是否清除会话（false=持久会话，true=临时会话）
     */
    private boolean cleanSession = false;

    /**
     * 默认QoS等级（0/1/2）
     */
    private int qos = 1;

    /**
     * 遗嘱主题（设备离线时自动发布）
     */
    private String willTopic;

    /**
     * 遗嘱消息（设备离线通知）
     */
    private String willMessage;

    /**
     * 遗嘱消息QoS
     */
    private int willQos = 1;

    /**
     * 遗嘱消息是否 retained
     */
    private boolean willRetained = true;
}

package com.example.mqtt.client;

import com.example.mqtt.config.MqttProperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * MQTT客户端工具类
 * 封装连接、订阅、发布、断开连接等操作
 */
@Slf4j
@Component
public class MqttClientUtil {
    private final MqttProperties mqttProperties;
    private MqttClient mqttClient;
    // 存储订阅的主题与回调关系
    private final ConcurrentMap<String, IMqttMessageListener> topicListeners = new ConcurrentHashMap<>();

    public MqttClientUtil(MqttProperties mqttProperties) {
        this.mqttProperties = mqttProperties;
    }

    /**
     * 初始化MQTT客户端并连接Broker
     */
    @PostConstruct
    public void init() {
        try {
            // 创建客户端（使用内存持久化，生产环境可改用文件持久化）
            mqttClient = new MqttClient(
                    mqttProperties.getBrokerUrl(),
                    mqttProperties.getClientId(),
                    new MemoryPersistence()
            );

            // 配置连接选项
            MqttConnectOptions connectOptions = createConnectOptions();

            // 设置全局回调（处理连接状态和消息）
            mqttClient.setCallback(new MqttGlobalCallback());

            // 连接Broker
            mqttClient.connect(connectOptions);
            log.info("MQTT客户端连接成功：clientId={}, brokerUrl={}",
                    mqttProperties.getClientId(), mqttProperties.getBrokerUrl());

            // 连接成功后发布上线消息
            publishOnlineStatus();

        } catch (MqttException e) {
            log.error("MQTT客户端初始化失败", e);
            throw new RuntimeException("MQTT客户端连接失败", e);
        }
    }

    /**
     * 创建连接选项
     */
    private MqttConnectOptions createConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();

        // 基本配置
        options.setCleanSession(mqttProperties.isCleanSession());
        options.setConnectionTimeout(mqttProperties.getConnectionTimeout());
        options.setKeepAliveInterval(mqttProperties.getKeepAliveInterval());

        // 认证配置（若Broker开启认证）
        if (Objects.nonNull(mqttProperties.getUsername()) && !mqttProperties.getUsername().isEmpty()) {
            options.setUserName(mqttProperties.getUsername());
            options.setPassword(mqttProperties.getPassword().toCharArray());
        }

        // 遗嘱消息配置（设备离线时自动发布）
        if (Objects.nonNull(mqttProperties.getWillTopic()) && !mqttProperties.getWillTopic().isEmpty()) {
            options.setWill(
                    mqttProperties.getWillTopic(),
                    mqttProperties.getWillMessage().getBytes(),
                    mqttProperties.getWillQos(),
                    mqttProperties.isWillRetained()
            );
        }

        // 启用自动重连（关键！保障弱网环境下的连接稳定性）
        options.setAutomaticReconnect(true);
        options.setMaxReconnectDelay(10000); // 最大重连延迟10秒

        return options;
    }

    /**
     * 订阅主题
     * @param topic 主题（支持通配符，如device/voice/+）
     * @param listener 消息监听器（处理该主题的消息）
     */
    public void subscribe(String topic, IMqttMessageListener listener) {
        try {
            if (!mqttClient.isConnected()) {
                log.warn("MQTT客户端未连接，订阅主题失败：{}", topic);
                return;
            }

            // 订阅主题并存储监听器
            mqttClient.subscribe(topic, mqttProperties.getQos(), listener);
            topicListeners.put(topic, listener);
            log.info("订阅主题成功：{}，QoS={}", topic, mqttProperties.getQos());
        } catch (MqttException e) {
            log.error("订阅主题失败：{}", topic, e);
        }
    }

    /**
     * 发布消息
     * @param topic 主题
     * @param message 消息内容
     * @return 是否发布成功
     */
    public boolean publish(String topic, String message) {
        return publish(topic, message, mqttProperties.getQos(), false);
    }

    /**
     * 发布消息（带QoS和retained参数）
     * @param topic 主题
     * @param message 消息内容
     * @param qos QoS等级
     * @param retained 是否保留消息
     * @return 是否发布成功
     */
    public boolean publish(String topic, String message, int qos, boolean retained) {
        try {
            if (!mqttClient.isConnected()) {
                log.warn("MQTT客户端未连接，发布消息失败：{} -> {}", topic, message);
                return false;
            }

            // 创建消息对象
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setPayload(message.getBytes());
            mqttMessage.setQos(qos);
            mqttMessage.setRetained(retained);

            // 发布消息
            mqttClient.publish(topic, mqttMessage);
            log.info("发布消息成功：{} -> {}（QoS={}, retained={}", topic, message, qos, retained);
            return true;
        } catch (MqttException e) {
            log.error("发布消息失败：{} -> {}", topic, message, e);
            return false;
        }
    }

    /**
     * 断开连接
     */
    @PreDestroy
    public void disconnect() {
        if (Objects.nonNull(mqttClient) && mqttClient.isConnected()) {
            try {
                // 发布离线消息（非必须，遗嘱消息会自动处理）
                publish(mqttProperties.getWillTopic(),
                        "{\"deviceId\":\"" + mqttProperties.getClientId() + "\",\"status\":\"offline\"}",
                        1, true);

                mqttClient.disconnect();
                log.info("MQTT客户端断开连接：{}", mqttProperties.getClientId());
            } catch (MqttException e) {
                log.error("MQTT客户端断开连接失败", e);
            }
        }
    }

    /**
     * 发布上线状态消息
     */
    private void publishOnlineStatus() {
        String onlineMessage = "{\"deviceId\":\"" + mqttProperties.getClientId() + "\",\"status\":\"online\",\"timestamp\":" + System.currentTimeMillis() + "}";
        publish(mqttProperties.getWillTopic(), onlineMessage, 1, true);
    }

    /**
     * MQTT全局回调类
     * 处理连接状态变化（连接丢失、重连成功等）
     */
    private class MqttGlobalCallback implements MqttCallback {
        /**
         * 连接丢失时触发
         */
        @Override
        public void connectionLost(Throwable cause) {
            log.warn("MQTT连接丢失，正在尝试重连...", cause);
        }

        /**
         * 收到消息时触发（仅全局消息，本文采用主题监听器模式，此处留空）
         */
        @Override
        public void messageArrived(String topic, MqttMessage message) {
            // 由主题对应的IMqttMessageListener处理，此处不重复处理
        }

        /**
         * 消息发布完成时触发
         */
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            try {
                if (token.isComplete()) {
                    log.debug("消息发布完成：{} -> {}", token.getTopics()[0], new String(token.getMessage().getPayload()));
                }
            } catch (MqttException e) {
                log.error("消息发布完成回调异常", e);
            }
        }
    }
}

package com.example.mqtt.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

/**
 * 语音设备消息处理器
 * 处理语音设备发布到device/voice/command主题的消息
 */
@Slf4j
@Component
public class VoiceDeviceMessageHandler implements IMqttMessageListener {

    /**
     * 处理收到的语音设备消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        String messageContent = new String(message.getPayload());
        log.info("收到语音设备消息：topic={}, message={}, QoS={}",
                topic, messageContent, message.getQos());

        try {
            // 解析JSON消息（假设格式：{"deviceId":"voice-123","command":"turn_on_light","timestamp":123456789}）
            JSONObject messageJson = JSON.parseObject(messageContent);

            // 提取设备ID和指令
            String deviceId = messageJson.getString("deviceId");
            String command = messageJson.getString("command");
            long timestamp = messageJson.getLongValue("timestamp");

            // 校验消息合法性
            if (deviceId == null || command == null) {
                log.error("语音设备消息格式错误，缺少deviceId或command：{}", messageContent);
                return;
            }

            // 根据指令类型处理（实际场景需调用业务服务）
            handleVoiceCommand(deviceId, command, timestamp);

        } catch (Exception e) {
            log.error("处理语音设备消息异常", e);
        }
    }

    /**
     * 处理语音指令
     */
    private void handleVoiceCommand(String deviceId, String command, long timestamp) {
        switch (command) {
            case "turn_on_light":
                log.info("执行指令：设备[{}]开启灯光，时间戳[{}]", deviceId, timestamp);
                // 调用灯光控制服务...
                break;
            case "turn_off_light":
                log.info("执行指令：设备[{}]关闭灯光，时间戳[{}]", deviceId, timestamp);
                // 调用灯光控制服务...
                break;
            case "set_volume":
                log.info("执行指令：设备[{}]调整音量，时间戳[{}]", deviceId, timestamp);
                // 调用音量控制服务...
                break;
            default:
                log.warn("未知语音指令：{}，设备ID：{}", command, deviceId);
        }
    }
}

package com.example.mqtt.config;

import com.example.mqtt.client.MqttClientUtil;
import com.example.mqtt.handler.VoiceDeviceMessageHandler;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * MQTT订阅配置类
 * 项目启动后自动订阅所需主题
 */
@Configuration
public class MqttSubscribeConfig {
    private final MqttClientUtil mqttClientUtil;
    private final VoiceDeviceMessageHandler voiceDeviceMessageHandler;

    public MqttSubscribeConfig(MqttClientUtil mqttClientUtil, VoiceDeviceMessageHandler voiceDeviceMessageHandler) {
        this.mqttClientUtil = mqttClientUtil;
        this.voiceDeviceMessageHandler = voiceDeviceMessageHandler;
    }

    /**
     * 订阅语音设备相关主题
     */
    @PostConstruct
    public void subscribeTopics() {
        // 订阅语音指令主题（支持通配符+，匹配所有设备的指令）
        mqttClientUtil.subscribe("device/voice/+/command", voiceDeviceMessageHandler);

        // 订阅设备状态主题（监控设备在线/离线）
        mqttClientUtil.subscribe("device/voice/status", (topic, message) -> {
            String status = new String(message.getPayload());
            log.info("收到设备状态更新：{} -> {}", topic, status);
            // 处理状态更新（如存储到数据库）...
        });

        // 订阅设备固件升级响应主题
        mqttClientUtil.subscribe("device/voice/+/firmware/response", (topic, message) -> {
            String response = new String(message.getPayload());
            log.info("收到固件升级响应：{} -> {}", topic, response);
            // 处理升级结果...
        });
    }
}

package com.example.mqtt.controller;

import com.example.mqtt.client.MqttClientUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;

/**
 * MQTT测试控制器
 * 提供API模拟向语音设备下发指令
 */
@Slf4j
@RestController
@RequestMapping("/api/voice")
@RequiredArgsConstructor
public class VoiceDeviceController {
    private final MqttClientUtil mqttClientUtil;

    /**
     * 向指定语音设备下发指令
     */
    @PostMapping("/{deviceId}/command")
    public String sendCommand(
            @PathVariable String deviceId,
            @RequestParam String command,
            @RequestParam(required = false, defaultValue = "1") int qos
    ) {
        // 主题格式：device/voice/{deviceId}/control
        String topic = "device/voice/" + deviceId + "/control";
        // 消息格式：{"command":"mute","timestamp":123456789,"from":"java-service"}
        String message = "{\"command\":\"" + command + "\",\"timestamp\":" + System.currentTimeMillis() + ",\"from\":\"java-service\"}";

        boolean success = mqttClientUtil.publish(topic, message, qos, false);
        return success ? "指令下发成功" : "指令下发失败";
    }

    /**
     * 查询设备在线状态
     */
    @GetMapping("/status")
    public String queryStatus() {
        // 实际场景应从数据库查询，此处简化
        return "Java服务在线，已连接MQTT Broker";
    }
}

package com.example.mqtt.service;

import com.example.mqtt.entity.DeviceStatus;
import com.example.mqtt.mapper.DeviceStatusMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;

/**
 * 设备状态管理服务
 * 存储和查询设备在线状态
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class DeviceStatusService {
    private final DeviceStatusMapper deviceStatusMapper;

    /**
     * 更新设备状态
     */
    public void updateStatus(String deviceId, String status, String reason) {
        DeviceStatus deviceStatus = new DeviceStatus();
        deviceStatus.setDeviceId(deviceId);
        deviceStatus.setStatus(status); // online/offline
        deviceStatus.setReason(reason);
        deviceStatus.setUpdateTime(LocalDateTime.now());

        // 存在则更新，不存在则插入
        if (deviceStatusMapper.existsByDeviceId(deviceId)) {
            deviceStatusMapper.updateByDeviceId(deviceStatus);
        } else {
            deviceStatusMapper.insert(deviceStatus);
        }
        log.info("更新设备状态：deviceId={}, status={}", deviceId, status);
    }

    /**
     * 查询设备状态
     */
    public DeviceStatus queryStatus(String deviceId) {
        return deviceStatusMapper.selectByDeviceId(deviceId);
    }
}

@Override
public void connectionLost(Throwable cause) {
    log.warn("MQTT连接丢失，正在尝试重连...", cause);
}

// 添加重连成功监听器（Paho客户端需通过扩展实现）
public void addReconnectListener(Runnable listener) {
    // 自定义重连监听器逻辑，定期检查连接状态
    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    executor.scheduleAtFixedRate(() -> {
        if (mqttClient.isConnected()) {
            listener.run();
            executor.shutdown(); // 执行一次后关闭
        }
    }, 1, 5, TimeUnit.SECONDS); // 每5秒检查一次
}

// 在初始化时添加重连后订阅逻辑
@PostConstruct
public void init() {
    // ... 原有连接逻辑 ...

    // 重连成功后重新订阅所有主题
    addReconnectListener(() -> {
        log.info("MQTT重连成功，重新订阅所有主题");
        topicListeners.forEach((topic, listener) -> {
            try {
                mqttClient.subscribe(topic, mqttProperties.getQos(), listener);
                log.info("重新订阅主题成功：{}", topic);
            } catch (MqttException e) {
                log.error("重新订阅主题失败：{}", topic, e);
            }
        });
        // 重连后发布上线消息
        publishOnlineStatus();
    });
}

/**
 * 带重试的消息发布
 */
public boolean publishWithRetry(String topic, String message, int maxRetry) {
    int retryCount = 0;
    while (retryCount < maxRetry) {
        if (publish(topic, message)) {
            return true;
        }
        retryCount++;
        log.warn("消息发布失败，重试第{}次：{} -> {}", retryCount, topic, message);
        try {
            // 指数退避重试（1s, 2s, 4s...）
            Thread.sleep(1000L * (1 << retryCount));
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }
    log.error("消息发布超过最大重试次数：{} -> {}", maxRetry, topic);
    return false;
}

// 使用示例：发布固件升级指令，最多重试3次
mqttClientUtil.publishWithRetry("device/voice/123/firmware", "upgrade", 3);

private MqttConnectOptions createConnectOptions() {
    MqttConnectOptions options = new MqttConnectOptions();
    // ... 其他配置 ...

    // 配置SSL
    try {
        // 加载CA证书
        InputStream caInput = new FileInputStream("src/main/resources/ssl/ca.crt");
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        X509Certificate caCert = (X509Certificate) cf.generateCertificate(caInput);

        // 初始化密钥库
        KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
        keyStore.load(null, null);
        keyStore.setCertificateEntry("ca-certificate", caCert);

        // 创建信任管理器工厂
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(
                TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(keyStore);

        // 初始化SSL上下文
        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
        sslContext.init(null, tmf.getTrustManagers(), new SecureRandom());
        options.setSocketFactory(sslContext.getSocketFactory());

    } catch (Exception e) {
        log.error("SSL配置失败", e);
        throw new RuntimeException("MQTT SSL配置失败", e);
    }

    return options;
}

// 在消息处理中捕获异常，避免单个消息处理失败影响整体
@Override
public void messageArrived(String topic, MqttMessage message) {
    try {
        // 消息处理逻辑...
    } catch (Exception e) {
        log.error("处理MQTT消息异常，topic={}", topic, e);
        // 可选：将异常消息存入死信队列，后续人工处理
        deadLetterService.save(topic, new String(message.getPayload()), e.getMessage());
    }
}

@Component
public class MqttHealthIndicator implements HealthIndicator {
    private final MqttClientUtil mqttClientUtil;

    @Override
    public Health health() {
        if (mqttClientUtil.isConnected()) {
            return Health.up()
                    .withDetail("clientId", mqttClientUtil.getClientId())
                    .withDetail("brokerUrl", mqttClientUtil.getBrokerUrl())
                    .build();
        } else {
            return Health.down()
                    .withDetail("reason", "MQTT客户端未连接")
                    .build();
        }
    }
}

@Component
public class BatchMessageHandler {
    private final DeviceStatusService statusService;
    private final BlockingQueue<DeviceStatus> messageQueue = new LinkedBlockingQueue<>(1000);
    private ScheduledExecutorService executor;

    @PostConstruct
    public void init() {
        // 每500ms批量处理一次
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(this::processBatch, 0, 500, TimeUnit.MILLISECONDS);
    }

    // 添加消息到队列
    public void addToBatch(DeviceStatus status) {
        try {
            messageQueue.put(status); // 队列满时阻塞等待
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("添加消息到批量队列失败", e);
        }
    }

    // 批量处理消息
    private void processBatch() {
        List<DeviceStatus> batch = new ArrayList<>(100);
        messageQueue.drainTo(batch, 100); // 一次最多取100条
        if (!batch.isEmpty()) {
            statusService.batchUpdate(batch);
            log.info("批量处理消息：{}条", batch.size());
        }
    }

    @PreDestroy
    public void shutdown() {
        executor.shutdown();
    }
}

@Configuration
public class ThreadPoolConfig {
    @Bean(name = "mqttMessageExecutor")
    public Executor mqttMessageExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(1000);
        executor.setThreadNamePrefix("mqtt-message-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

// 在消息处理器中使用
@Async("mqttMessageExecutor")
@Override
public void messageArrived(String topic, MqttMessage message) {
    // 异步处理消息...
}