Java 开发 Spring Boot 集成 MQTT 消息协议实现消息推送与订阅功能详解 - 主机乐

<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>

2. 配置MQTT连接信息

在application.properties中配置MQTT服务器地址和认证信息：

代码语言：properties

AI代码解释

复制

# MQTT配置
mqtt.host=tcp://localhost:1883
mqtt.clientId=spring-boot-mqtt-client
mqtt.username=admin
mqtt.password=password
mqtt.defaultTopic=test/topic

3. 创建MQTT配置类

配置MQTT客户端工厂和消息通道：

代码语言：java

AI代码解释

复制

@Configuration
public class MqttConfig {

    @Value("${mqtt.host}")
    private String host;
    
    @Value("${mqtt.clientId}")
    private String clientId;
    
    @Value("${mqtt.username}")
    private String username;
    
    @Value("${mqtt.password}")
    private String password;
    
    @Value("${mqtt.defaultTopic}")
    private String defaultTopic;

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

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

4. 实现消息发布服务

创建服务类实现消息发布功能：

代码语言：java

AI代码解释

复制

@Service
public class MqttPublisher {

    private final MqttPahoClientFactory mqttClientFactory;
    private final String clientId;
    
    public MqttPublisher(MqttPahoClientFactory mqttClientFactory, 
                         @Value("${mqtt.clientId}") String clientId) {
        this.mqttClientFactory = mqttClientFactory;
        this.clientId = clientId + "-publisher";
    }
    
    public void publish(String topic, String payload) {
        try (MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                clientId, mqttClientFactory)) {
            messageHandler.setAsync(true);
            messageHandler.setDefaultTopic(topic);
            messageHandler.start();
            messageHandler.handleMessage(new GenericMessage<>(payload));
        } catch (Exception e) {
            log.error("MQTT publish error: {}", e.getMessage(), e);
        }
    }
}

5. 实现消息订阅服务

配置消息监听器接收订阅消息：

代码语言：java

AI代码解释

复制

@Configuration
public class MqttSubscriberConfig {

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound(MqttPahoClientFactory mqttClientFactory) {
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(
                        "subscriberClient", 
                        mqttClientFactory,
                        "test/topic/#"); // 订阅主题通配符
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
            String payload = message.getPayload().toString();
            System.out.println("收到MQTT消息 - 主题: " + topic + ", 内容: " + payload);
            // 处理接收到的消息
        };
    }
}

三、应用实例：物联网设备监控系统

场景说明：

构建一个简单的物联网设备监控系统，实现：

设备状态数据定时上报
服务器远程控制指令下发
实时数据展示与告警

1. 设备端实现（模拟）

代码语言：java

AI代码解释

复制

@Service
public class DeviceSimulator {

    @Autowired
    private MqttPublisher mqttPublisher;
    
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    
    public void startReporting(String deviceId) {
        scheduler.scheduleAtFixedRate(() -> {
            // 生成模拟设备数据
            Map<String, Object> payload = new HashMap<>();
            payload.put("deviceId", deviceId);
            payload.put("temperature", 20 + new Random().nextInt(10));
            payload.put("humidity", 40 + new Random().nextInt(20));
            payload.put("status", "online");
            payload.put("timestamp", System.currentTimeMillis());
            
            // 发布设备数据到MQTT
            String jsonPayload = new ObjectMapper().writeValueAsString(payload);
            mqttPublisher.publish("device/data/" + deviceId, jsonPayload);
        }, 0, 5, TimeUnit.SECONDS); // 每5秒上报一次
    }
}

2. 服务端数据处理

代码语言：java

AI代码解释

复制

@Service
public class DeviceDataService {

    @Autowired
    private DeviceRepository deviceRepository;
    
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public void handleDeviceData(Message<?> message) {
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
        String payload = message.getPayload().toString();
        
        // 解析设备数据
        if (topic.startsWith("device/data/")) {
            try {
                DeviceData deviceData = new ObjectMapper().readValue(payload, DeviceData.class);
                // 保存设备数据
                deviceRepository.save(deviceData);
                // 检查告警阈值
                checkAlarms(deviceData);
            } catch (JsonProcessingException e) {
                log.error("解析设备数据失败: {}", e.getMessage(), e);
            }
        }
    }
    
    private void checkAlarms(DeviceData data) {
        // 检查温度是否超过阈值
        if (data.getTemperature() > 30) {
            // 触发高温告警
            sendAlarm("高温告警", 
                    "设备 " + data.getDeviceId() + " 温度异常: " + data.getTemperature() + "°C");
        }
    }
    
    private void sendAlarm(String title, String content) {
        // 发送告警通知
        log.warn("ALARM: {} - {}", title, content);
    }
}

3. 控制指令下发

代码语言：java

AI代码解释

复制

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

    @Autowired
    private MqttPublisher mqttPublisher;
    
    @PostMapping("/{deviceId}/command")
    public ResponseEntity<String> sendCommand(
            @PathVariable String deviceId,
            @RequestBody DeviceCommand command) {
        
        try {
            // 将命令转换为JSON格式
            String payload = new ObjectMapper().writeValueAsString(command);
            // 发布命令到设备
            mqttPublisher.publish("device/command/" + deviceId, payload);
            return ResponseEntity.ok("命令已发送");
        } catch (Exception e) {
            log.error("发送设备命令失败: {}", e.getMessage(), e);
            return ResponseEntity.status(500).body("发送命令失败");
        }
    }
}

四、测试与验证

1. 单元测试示例

代码语言：java

AI代码解释

复制

@SpringBootTest
class MqttIntegrationTest {

    @Autowired
    private MqttPublisher mqttPublisher;
    
    @Test
    void testMqttPublishAndSubscribe() throws Exception {
        // 创建测试主题
        String testTopic = "test/integration/" + UUID.randomUUID().toString();
        String testPayload = "Hello, MQTT!";
        
        // 设置异步测试监听器
        CountDownLatch latch = new CountDownLatch(1);
        AtomicReference<String> receivedPayload = new AtomicReference<>();
        
        MessageHandler testHandler = message -> {
            receivedPayload.set(message.getPayload().toString());
            latch.countDown();
        };
        
        // 发布消息
        mqttPublisher.publish(testTopic, testPayload);
        
        // 等待消息接收
        assertTrue(latch.await(5, TimeUnit.SECONDS));
        assertEquals(testPayload, receivedPayload.get());
    }
}