Java 中 Spring Boot 3.2 集成 MQTT 5.0 实现消息推送与订阅功能的详细技术方案

原创

啦啦啦191

发布于 2025-06-10 10:10:26

4950

文章被收录于专栏：Java开发Java开发

Spring Boot 3.2集成MQTT 5.0实现消息推送与订阅技术方案

一、技术选型与架构设计

1. 核心技术栈

Spring Boot 3.2.0 (基于Java 17)
Eclipse Paho MQTT Client 1.2.5
MQTT 5.0 协议 (支持属性扩展、增强的错误处理)
HiveMQ (开源MQTT Broker)
WebSocket 支持 (可选)

2. 架构图

+------------------+     +------------------+     +------------------+
|  设备/前端应用   |<--->|   MQTT Broker    |<--->|  Spring Boot应用  |
+------------------+     +------------------+     +------------------+
      发布/订阅              消息路由               业务处理

二、项目搭建与配置

1. 创建Spring Boot项目

使用Spring Initializr创建项目，添加以下依赖：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-mqtt</artifactId>
    </dependency>
    <dependency>
        <groupId>com.hivemq</groupId>
        <artifactId>hivemq-mqtt-client</artifactId>
        <version>1.3.0</version>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

2. 配置MQTT连接

使用HiveMQ客户端实现MQTT 5.0支持：

@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 Mqtt5AsyncClient mqttClient() {
        Mqtt5AsyncClient client = MqttClient.builder()
            .useMqttVersion5()
            .identifier(clientId)
            .serverHost(brokerUrl)
            .serverPort(1883)
            .buildAsync();
        
        // 添加认证信息
        Mqtt5ConnectBuilder.Mqtt5ConnectWithUserPropertiesBuilder connectBuilder = 
            Mqtt5ClientConnectionConfig.builder()
                .automaticReconnectWithDefaultConfig()
                .build();
        
        return client;
    }
    
    // 配置消息转换器和质量服务
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{brokerUrl});
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        options.setMqttVersion(MqttConnectOptions.MQTT_VERSION_5);
        factory.setConnectionOptions(options);
        return factory;
    }
}

3. 配置文件示例

# MQTT配置
mqtt.broker-url=tcp://localhost:1883
mqtt.client-id=spring-boot-mqtt-client
mqtt.username=admin
mqtt.password=password
mqtt.default-qos=1
mqtt.keep-alive-interval=30

三、消息发布服务实现

1. 通用消息发布服务

@Service
public class MqttPublisherService {

    private static final Logger logger = LoggerFactory.getLogger(MqttPublisherService.class);
    
    private final Mqtt5AsyncClient mqttClient;
    
    @Autowired
    public MqttPublisherService(Mqtt5AsyncClient mqttClient) {
        this.mqttClient = mqttClient;
    }
    
    /**
     * 发布MQTT消息
     * @param topic 主题
     * @param payload 消息内容
     * @param qos 服务质量等级
     * @param retained 是否保留消息
     */
    public CompletableFuture<Void> publish(String topic, String payload, int qos, boolean retained) {
        Mqtt5Publish publishMessage = Mqtt5Publish.builder()
            .topic(topic)
            .payload(ByteBuffer.wrap(payload.getBytes(StandardCharsets.UTF_8)))
            .qos(MqttQos.fromCode(qos))
            .retain(retained)
            .build();
        
        return mqttClient.publish(publishMessage)
            .thenAccept(publishResult -> logger.info("消息发布成功: {}", publishResult))
            .exceptionally(ex -> {
                logger.error("消息发布失败: {}", ex.getMessage(), ex);
                return null;
            });
    }
    
    // 重载方法，使用默认QoS和retained设置
    public CompletableFuture<Void> publish(String topic, String payload) {
        return publish(topic, payload, 1, false);
    }
}

2. 领域特定消息发布示例

@Service
public class DeviceMessageService {
    
    private static final String DEVICE_DATA_TOPIC = "v1/devices/me/telemetry";
    
    @Autowired
    private MqttPublisherService publisherService;
    
    public CompletableFuture<Void> sendDeviceData(String deviceId, Map<String, Object> data) {
        try {
            ObjectMapper mapper = new ObjectMapper();
            String payload = mapper.writeValueAsString(data);
            String topic = String.format("%s/%s", DEVICE_DATA_TOPIC, deviceId);
            
            return publisherService.publish(topic, payload);
        } catch (JsonProcessingException e) {
            logger.error("序列化设备数据失败: {}", e.getMessage(), e);
            return CompletableFuture.failedFuture(e);
        }
    }
}

四、消息订阅服务实现

1. 基于注解的消息处理

@Component
public class MqttMessageListener {

    private static final Logger logger = LoggerFactory.getLogger(MqttMessageListener.class);
    
    @Autowired
    private DeviceService deviceService;
    
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleMessage(Message<?> message) {
        String topic = (String) message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC);
        String payload = new String((byte[]) message.getPayload(), StandardCharsets.UTF_8);
        
        logger.info("收到MQTT消息 - 主题: {}, 内容: {}", topic, payload);
        
        // 根据主题路由消息处理
        if (topic.startsWith("v1/devices/")) {
            handleDeviceMessage(topic, payload);
        } else if (topic.startsWith("system/")) {
            handleSystemMessage(topic, payload);
        }
    }
    
    private void handleDeviceMessage(String topic, String payload) {
        try {
            // 解析设备ID
            String deviceId = topic.split("/")[2];
            
            // 解析JSON数据
            ObjectMapper mapper = new ObjectMapper();
            JsonNode data = mapper.readTree(payload);
            
            // 处理设备数据
            deviceService.processDeviceData(deviceId, data);
        } catch (Exception e) {
            logger.error("处理设备消息失败: {}", e.getMessage(), e);
        }
    }
    
    private void handleSystemMessage(String topic, String payload) {
        // 处理系统消息逻辑
    }
}

2. 配置消息订阅

@Configuration
public class MqttSubscriberConfig {

    @Value("${mqtt.client-id}")
    private String clientId;
    
    @Autowired
    private MqttPahoClientFactory mqttClientFactory;
    
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
    
    @Bean
    public MessageProducer inbound() {
        MqttPahoMessageDrivenChannelAdapter adapter = 
            new MqttPahoMessageDrivenChannelAdapter(
                clientId + "-subscriber", 
                mqttClientFactory,
                "v1/devices/#",  // 订阅设备相关主题
                "system/#"       // 订阅系统相关主题
            );
        
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
    
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                // 消息将被路由到MqttMessageListener
            }
        };
    }
}

五、应用实例：智能家居控制系统

1. 设备状态监控与控制

设备数据模型

public record DeviceData(
    String deviceId,
    double temperature,
    double humidity,
    boolean powerStatus,
    LocalDateTime timestamp
) {}

设备服务实现

@Service
public class SmartHomeService {
    
    private static final String CONTROL_TOPIC = "v1/devices/me/control";
    
    @Autowired
    private MqttPublisherService publisherService;
    
    @Autowired
    private DeviceRepository deviceRepository;
    
    // 处理设备上报数据
    public void processDeviceData(String deviceId, JsonNode data) {
        // 解析数据
        double temperature = data.path("temperature").asDouble();
        double humidity = data.path("humidity").asDouble();
        boolean powerStatus = data.path("powerStatus").asBoolean();
        
        // 创建设备数据对象
        DeviceData deviceData = new DeviceData(
            deviceId,
            temperature,
            humidity,
            powerStatus,
            LocalDateTime.now()
        );
        
        // 保存数据
        deviceRepository.save(deviceData);
        
        // 检查自动化规则
        checkAutomationRules(deviceData);
    }
    
    // 发送控制命令到设备
    public CompletableFuture<Void> sendDeviceCommand(String deviceId, String command) {
        String topic = String.format("%s/%s", CONTROL_TOPIC, deviceId);
        return publisherService.publish(topic, command);
    }
    
    // 自动化规则检查
    private void checkAutomationRules(DeviceData data) {
        // 示例：温度超过30度时自动打开空调
        if (data.deviceId().endsWith("thermostat") && data.temperature() > 30) {
            sendDeviceCommand("air-conditioner-01", "{\"command\":\"ON\"}");
        }
    }
}

2. REST API实现

@RestController
@RequestMapping("/api/v1/devices")
public class DeviceController {

    @Autowired
    private SmartHomeService smartHomeService;
    
    @PostMapping("/{deviceId}/command")
    public ResponseEntity<?> sendCommand(
        @PathVariable String deviceId,
        @RequestBody String command
    ) {
        smartHomeService.sendDeviceCommand(deviceId, command)
            .thenAccept(v -> ResponseEntity.ok().build())
            .exceptionally(ex -> ResponseEntity.status(500).body(ex.getMessage()));
        
        return ResponseEntity.accepted().build();
    }
    
    @GetMapping("/{deviceId}/data")
    public ResponseEntity<DeviceData> getDeviceData(@PathVariable String deviceId) {
        Optional<DeviceData> deviceData = smartHomeService.getLatestData(deviceId);
        return deviceData.map(ResponseEntity::ok)
                         .orElse(ResponseEntity.notFound().build());
    }
}

六、安全增强与性能优化

1. TLS/SSL配置

@Bean
public MqttConnectOptions mqttConnectOptions() {
    MqttConnectOptions options = new MqttConnectOptions();
    options.setServerURIs(new String[]{brokerUrl});
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    
    // 配置TLS
    if (useTls) {
        try {
            SSLSocketFactory sslSocketFactory = createSSLSocketFactory();
            options.setSocketFactory(sslSocketFactory);
        } catch (Exception e) {
            logger.error("配置TLS连接失败: {}", e.getMessage(), e);
        }
    }
    
    return options;
}

private SSLSocketFactory createSSLSocketFactory() throws Exception {
    // 加载证书
    KeyStore keyStore = KeyStore.getInstance("JKS");
    InputStream inputStream = getClass().getResourceAsStream("/client.jks");
    keyStore.load(inputStream, "password".toCharArray());
    
    // 初始化SSL上下文
    SSLContext sslContext = SSLContext.getInstance("TLS");
    TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(
        TrustManagerFactory.getDefaultAlgorithm());
    trustManagerFactory.init(keyStore);
    sslContext.init(null, trustManagerFactory.getTrustManagers(), new SecureRandom());
    
    return sslContext.getSocketFactory();
}

2. 异步处理与背压控制

@Service
public class AsyncMessageProcessor {
    
    private final ExecutorService threadPool = Executors.newFixedThreadPool(10);
    
    public void processMessageAsync(String topic, String payload) {
        CompletableFuture.runAsync(() -> {
            try {
                // 处理消息的业务逻辑
                processMessage(topic, payload);
            } catch (Exception e) {
                logger.error("异步处理消息失败: {}", e.getMessage(), e);
            }
        }, threadPool);
    }
    
    private void processMessage(String topic, String payload) {
        // 消息处理逻辑
    }
}

七、测试与监控

1. 单元测试示例

@SpringBootTest
@ActiveProfiles("test")
class MqttIntegrationTest {

    @Autowired
    private MqttPublisherService publisherService;
    
    @Autowired
    private TestMessageCollector messageCollector;
    
    @Test
    void testPublishAndSubscribe() throws Exception {
        String testTopic = "test/unit/" + UUID.randomUUID().toString();
        String testPayload = "Test Message";
        
        // 设置预期消息
        messageCollector.expectMessage(testTopic, testPayload);
        
        // 发布消息
        publisherService.publish(testTopic, testPayload).get(5, TimeUnit.SECONDS);
        
        // 验证消息是否收到
        assertTrue(messageCollector.waitForMessage(5, TimeUnit.SECONDS));
    }
}

2. 监控指标

使用Micrometer添加MQTT相关监控指标：

@Bean
public MeterRegistryCustomizer<MeterRegistry> configurer(
        @Value("${spring.application.name}") String applicationName) {
    return (registry) -> registry.config().commonTags("application", applicationName);
}

// 在消息处理中添加计数器
@Service
public class MqttMetricsService {
    
    private final Counter publishCounter;
    private final Counter subscribeCounter;
    private final Timer messageProcessingTimer;
    
    public MqttMetricsService(MeterRegistry registry) {
        publishCounter = registry.counter("mqtt.publish.count");
        subscribeCounter = registry.counter("mqtt.subscribe.count");
        messageProcessingTimer = registry.timer("mqtt.message.processing.time");
    }
    
    public void incrementPublishCount() {
        publishCounter.increment();
    }
    
    public void incrementSubscribeCount() {
        subscribeCounter.increment();
    }
    
    public <T> T recordProcessingTime(Supplier<T> operation) {
        return messageProcessingTimer.record(operation);
    }
}

八、生产环境部署

1. MQTT Broker选型

HiveMQ CE：开源、高性能、支持MQTT 5.0
Mosquitto：轻量级、易于部署
EMQ X：企业级、支持百万级连接

2. Docker部署示例

version: '3'
services:
  mqtt-broker:
    image: hivemq/hivemq-ce
    ports:
      - "1883:1883"  # MQTT
      - "8080:8080"  # HiveMQ Web UI
    volumes:
      - ./hivemq/config:/opt/hivemq/conf
      - ./hivemq/data:/opt/hivemq/data
      - ./hivemq/log:/opt/hivemq/log
    restart: always
    
  spring-boot-app:
    build: .
    ports:
      - "8081:8081"
    environment:
      - MQTT_BROKER_URL=mqtt-broker
      - MQTT_USERNAME=admin
      - MQTT_PASSWORD=password
    depends_on:
      - mqtt-broker
    restart: always