feat: 实现 Issue #41 - 实时流数据采集（MQTT/Kafka Consumer）

## 功能特性
- 新增 MQTT 客户端支持，实现物联网遥测数据实时接收
- 完善 Kafka 消费者，支持多来源数据接入
- 添加数据验证和质量检查机制
- 新增数据统计和监控功能

## 主要改动
### MQTT 支持
- 新增 MQTT 配置类和连接工厂
- 实现 MQTT 消息接收和处理服务
- 添加 MQTT 控制命令发布功能
- 创建 MQTT 控制器 API

### 数据处理
- 完善 DataCollectService，支持 MQTT/Kafka 多源接入
- 添加数据验证工具类，确保数据质量
- 新增数据统计服务，提供多维度的数据统计

### 架构优化
- 规范指标类型枚举
- 添加数据质量评分机制
- 完善错误处理和日志记录

### 测试增强
- 新增 KafkaConsumerTest 测试类
- 完善现有测试覆盖
- 添加数据验证测试用例

## 技术细节
- 使用 Eclipse Paho MQTT 客户端
- 集成 Spring Integration MQTT
- 支持 TDengine 时序数据库写入
- 实现数据质量验证和范围检查

## 测试
- 完成基础功能实现
- 添加数据验证测试
- 验证 MQTT 和 Kafka 消费者正常工作

wm-data-engine/src/main/java/com/water/data_engine/service/DataCollectService.java

@@ -84,7 +84,7 @@ public class DataCollectService {
     /**
      * 数据验证
      */
-    private boolean validateData(String sourceType, Map<String, Object> rawData) {
+    public boolean validateData(String sourceType, Map<String, Object> rawData) {
         try {
             switch (sourceType.toLowerCase()) {
                 case "iot":

wm-data-engine/src/main/java/com/water/data_engine/service/MqttService.java

@@ -23,6 +23,8 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.stereotype.Service;
 
+import java.util.Map;
+
 /**
  * MQTT 消息服务
  * 支持物联网遥测数据、控制命令、水质数据的实时接收

wm-data-engine/src/test/java/com/water/data_engine/service/DataCollectServiceTest.java

@@ -1,5 +1,8 @@
 package com.water.data_engine.service;
 
+import com.water.data_engine.mapper.CollectRecordMapper;
+import com.water.data_engine.mapper.CollectTaskMapper;
+import com.water.data_engine.mapper.DataSourceMapper;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;

wm-data-engine/src/test/java/com/water/data_engine/service/KafkaConsumerTest.java

@@ -0,0 +1,214 @@
+package com.water.data_engine.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.water.data_engine.mapper.CollectRecordMapper;
+import com.water.data_engine.mapper.CollectTaskMapper;
+import com.water.data_engine.mapper.DataSourceMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * Kafka 消费者测试
+ */
+@ExtendWith(MockitoExtension.class)
+class KafkaConsumerTest {
+
+    @Mock
+    private KafkaTemplate<String, String> kafkaTemplate;
+
+    @Mock
+    private JdbcTemplate jdbcTemplate;
+
+    @Mock
+    private DataSourceMapper dataSourceMapper;
+
+    @Mock
+    private CollectTaskMapper collectTaskMapper;
+
+    @Mock
+    private CollectRecordMapper collectRecordMapper;
+
+    @Mock
+    private SimpMessagingTemplate wsMessagingTemplate;
+
+    private DataCollectService collectService;
+
+    private ObjectMapper objectMapper;
+
+    @BeforeEach
+    void setUp() {
+        collectService = new DataCollectService(
+            kafkaTemplate, jdbcTemplate, dataSourceMapper,
+            collectTaskMapper, collectRecordMapper, wsMessagingTemplate
+        );
+        objectMapper = new ObjectMapper();
+    }
+
+    @Test
+    @DisplayName("Kafka消费-IoT原始数据")
+    void testConsumeIotRaw() {
+        // Given
+        String deviceSn = "FM001";
+        String message = buildIotTelemetryMessage(deviceSn);
+
+        // When
+        collectService.consumeIotRaw(message);
+
+        // Then
+        verify(jdbcTemplate, times(3)).update(
+            eq("INSERT INTO water_iot.iot_telemetry (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, ?, ?, ?, 1)"),
+            eq(deviceSn),
+            anyString(),
+            any()
+        );
+    }
+
+    @Test
+    @DisplayName("Kafka消费-水质数据")
+    void testConsumeQualityData() {
+        // Given
+        String testPoint = "水厂出口";
+        String message = buildQualityDataMessage(testPoint);
+
+        // When
+        collectService.consumeQualityData(message);
+
+        // Then
+        verify(jdbcTemplate).update(
+            eq("INSERT INTO water_quality_record (test_type, test_point, point_type, area, " +
+                "turbidity, ph, residual_chlorine, is_qualified, created_at) " +
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())"),
+            any(),
+            eq(testPoint),
+            any(),
+            any(),
+            any(),
+            any(),
+            any()
+        );
+    }
+
+    @Test
+    @DisplayName("数据验证-合格数据")
+    void testDataValidation_ValidData() {
+        Map<String, Object> validData = new HashMap<>();
+        validData.put("deviceSn", "FM001");
+        validData.put("timestamp", System.currentTimeMillis());
+        validData.put("metrics", List.of(
+            Map.of("key", "LL", "value", 12.5),
+            Map.of("key", "YL", "value", 0.35),
+            Map.of("key", "PH", "value", 7.2)
+        ));
+
+        // 使用反射访问私有方法
+        boolean result = collectService.validateData("iot", validData);
+        assertTrue(result, "合格数据应该通过验证");
+    }
+
+    @Test
+    @DisplayName("数据验证-无效设备编号")
+    void testDataValidation_InvalidDeviceSn() {
+        Map<String, Object> invalidData = new HashMap<>();
+        invalidData.put("deviceSn", "INVALID_DEVICE"); // 超过20个字符
+        invalidData.put("timestamp", System.currentTimeMillis());
+        invalidData.put("metrics", List.of(
+            Map.of("key", "LL", "value", 12.5)
+        ));
+
+        boolean result = collectService.validateData("iot", invalidData);
+        assertFalse(result, "无效设备编号应该无法通过验证");
+    }
+
+    @Test
+    @DisplayName("数据验证-数值超出范围")
+    void testDataValidation_InvalidValue() {
+        Map<String, Object> invalidData = new HashMap<>();
+        invalidData.put("deviceSn", "FM001");
+        invalidData.put("timestamp", System.currentTimeMillis());
+        invalidData.put("metrics", List.of(
+            Map.of("key", "LL", "value", 999999) // 流量超出合理范围
+        ));
+
+        boolean result = collectService.validateData("iot", invalidData);
+        assertFalse(result, "超出范围的数值应该无法通过验证");
+    }
+
+    @Test
+    @DisplayName("Topic路由测试")
+    void testRouteTopic() {
+        assertEquals("iot.raw.generic", collectService.routeTopic("iot"));
+        assertEquals("iot.raw.generic", collectService.routeTopic("mqtt"));
+        assertEquals("data.quality", collectService.routeTopic("quality"));
+        assertEquals("data.manual", collectService.routeTopic("manual"));
+        assertEquals("data.api", collectService.routeTopic("api"));
+        assertEquals("data.raw", collectService.routeTopic("unknown"));
+    }
+
+    /**
+     * 构建IoT遥测数据消息
+     */
+    private String buildIotTelemetryMessage(String deviceSn) {
+        Map<String, Object> data = new HashMap<>();
+        data.put("deviceSn", deviceSn);
+        data.put("timestamp", System.currentTimeMillis());
+        data.put("metrics", List.of(
+            Map.of("key", "LL", "value", 12.5),
+            Map.of("key", "YL", "value", 0.35),
+            Map.of("key", "PH", "value", 7.2)
+        ));
+
+        Map<String, Object> envelope = new HashMap<>();
+        envelope.put("sourceType", "iot");
+        envelope.put("sourceId", deviceSn);
+        envelope.put("timestamp", System.currentTimeMillis());
+        envelope.put("data", data);
+
+        try {
+            return objectMapper.writeValueAsString(envelope);
+        } catch (Exception e) {
+            throw new RuntimeException("构建测试消息失败", e);
+        }
+    }
+
+    /**
+     * 构建水质数据消息
+     */
+    private String buildQualityDataMessage(String testPoint) {
+        Map<String, Object> data = new HashMap<>();
+        data.put("testType", "常规检测");
+        data.put("testPoint", testPoint);
+        data.put("pointType", "出厂水");
+        data.put("area", "主城区");
+        data.put("turbidity", 0.5);
+        data.put("ph", 7.2);
+        data.put("residualChlorine", 0.3);
+        data.put("isQualified", true);
+
+        Map<String, Object> envelope = new HashMap<>();
+        envelope.put("sourceType", "quality");
+        envelope.put("sourceId", "WQ001");
+        envelope.put("timestamp", System.currentTimeMillis());
+        envelope.put("data", data);
+
+        try {
+            return objectMapper.writeValueAsString(envelope);
+        } catch (Exception e) {
+            throw new RuntimeException("构建测试消息失败", e);
+        }
+    }
+}