Java 大数据在智能家居环境监测与智能调节中的应用

看到'✅ Drools 决策规则容器初始化完成'说明成功。

提交 Spark Streaming 任务（本地模式）：

cd streaming-process-module mvn clean package spark-submit --class com.smarthome.streaming.SmartHomeStreamProcessor --master local[*] target/spark-streaming-processor-1.0.0.jar 

本地模式用local[*]，表示用所有 CPU 核心；

运行数据采集模块：

cd data-collect-module mvn clean package java -jar target/data-collect-module-1.0.0.jar 

看到'✅ MQTT 数据采集客户端初始化完成'说明成功；

Step 3：效果验证（用 Postman 模拟数据）

• 查看 Spark 处理结果：Spark 控制台会打印房间指标，如'room1|avg_temperature:28.0|avg_humidity:35.0|anomaly_level:medium'；
• 查看决策结果：决策引擎日志会显示'规则触发 [加湿]：房间 room1 湿度过低（当前 35.0%），加湿器开启至 50%'。

模拟设备数据：发送 POST 请求到 EMQX 的http://localhost:8083/api/v5/mqtt/publish（EMQX API），Body 如下：

{"topic":"smarthome/device/sensor/room1","payload":"{\"device_id\":\"room1_sensor01\",\"temp\":28,\"hum\":35,\"illum\":500}","qos":1}

headers 加Authorization：Basic YWRtaW46cGFzc3dvcmQ=（默认账号 admin，密码 public）；

到这里，你就成功跑通了'设备数据采集→实时处理→智能决策'的完整流程！接下来可以继续看正文的详细技术细节和项目案例。

正文：

智能家居的核心是'感知 - 分析 - 决策 - 执行'的闭环，而 Java 大数据正是打通这个闭环的'神经中枢'。下文会先拆解智能家居的真实痛点，详解 Java 技术栈选型的底层逻辑，再提供可直接复用的核心模块代码（附完整依赖和部署步骤），最后通过高端小区案例展示落地效果，并拓展 AI 学习、跨空间联动等未来方向。

一、智能家居环境监测与调节的核心痛点

在做第一个智能家居项目时，用 2 周时间访谈了 50 位用户和 10 家设备厂商，总结出 3 类阻碍'真智能'的核心痛点，这些痛点也成了后续技术选型的'指挥棒'。

1.1 设备数据的'异构化'困境

智能家居设备来自不同厂商，数据格式、传输协议差异极大，就像'说着不同语言的士兵'，根本无法协同作战。

1.1.1 多源数据的'协议壁垒'

至今记得第一个项目上线前的紧急调试：某批空调的'温度'字段是'temp_val'，而我们代码里写的是'temperature'，导致大屏显示'null'，连夜改代码加字段映射才解决——这也是后来必须做'数据标准化'的直接原因。

1.1.2 数据规模的'爆发式增长'

以一个 3 室 2 厅的家庭为例，若部署 15 个智能设备（空调×2、净化器×1、温湿度传感器×6、窗帘×2、智能插座×4），数据规模如下：

• 实时数据：日均 100MB（主要是传感器和空调数据）；
• 历史数据：按保存 1 年计算，约 36GB（含环境指标、用户操作记录）；
• 结构化数据：用户配置、设备属性等约 50MB。

传统的单机数据库（如 SQLite）根本扛不住——第二个项目中，某小区物业反映'环境监测大屏延迟超 30 秒'，查下来是 SQLite 单表数据超 1000 万条，查询一次要 25 秒，完全失去了实时调节的意义。

1.2 实时调节的'滞后性'痛点

1.2.1 决策响应的'秒级差距'

用户回家前 10 分钟，想让空调自动调到 24℃，但传统智能家居需要手动在 APP 上预约——这是'被动响应'；而理想的状态是：系统通过手机定位 + 历史行为分析，提前 10 分钟启动空调，用户进门就能享受到'刚刚好'的温度——这需要'秒级决策'。

第二个项目中，初期用'Java 定时器 + MySQL 查询'做决策，响应延迟 8 秒，用户反馈'刚进门还是热的'；后来换成 Spark Streaming，把延迟降到 2 秒内，用户满意度立刻提升了 30%。

1.2.2 调节策略的'经验主义'

传统智能家居的调节策略靠'固定阈值'（如温度 > 26℃开空调），但忽略了'环境联动'和'用户习惯'。比如：

• 同样是 26℃，湿度 60%（闷热）和湿度 30%（干爽）的体感完全不同，但传统系统只会按温度阈值操作；
• 用户 A（老人）喜欢睡前开 25℃，用户 B（年轻人）喜欢开 23℃，但系统给所有人都设 24℃，老人觉得冷，年轻人觉得热。

1.3 隐私安全的'敏感性'挑战

环境监测涉及大量用户隐私数据（如'几点回家''几点睡觉''房间有人没人'），但很多智能家居平台存在'数据明文传输''存储不安全'的问题。

第三个项目做安全测试时，我们用抓包工具轻易就获取到了某品牌温湿度传感器的明文数据——从'晚上 10 点湿度突然下降'能推断出'用户开了空调，可能准备睡觉了'；从'周末上午 9 点房间有光照变化'能推断出'用户起床拉开了窗帘'。这些数据一旦泄露，后果不堪设想——这也是后来所有数据传输都加 SSL 加密、存储用脱敏的原因。

二、Java 大数据技术栈的选型逻辑

技术选型不是'选最酷的'，而是'选最能解决问题的'。在 3 个项目中对比了多套技术栈，最终确定了以 Java 为核心的方案，下面详解选型逻辑（含当时团队的争论和决策过程）。

2.1 技术栈选型对比与决策

当时的选型争论：团队里有个年轻工程师坚持用 Flink，说'流处理更先进'，但我算了一笔账：Flink 的运维需要额外学 Checkpoint 配置、状态后端管理，团队要花 1 个月学习，而 Spark Streaming 我们熟，2 周就能开发完；而且智能家居的实时需求是'5 秒内响应'，Spark 的微批处理完全能满足——技术选型要'量力而行'，不是越先进越好。

2.2 系统整体架构

下面的架构图是根据第三个项目（高端小区，500 户）绘制的，采用纵向布局，每个模块加了业务图标和详细说明，节点 padding 调大至 15px，确保文字不拥挤，颜色按'感知 - 传输 - 处理 - 存储 - 应用 - 执行'分层，清晰易读：

架构设计的核心思考：在设计时重点考虑了'扩展性'和'容错性'。比如传输层用 EMQX 作为 MQTT Broker，支持百万级设备连接，后续小区扩容到 2000 户也不用换架构；存储层用 Cassandra 存时序数据，因为它按时间分区，查'近 24 小时温度'比 MySQL 快 10 倍——第三个项目中，这个架构支撑了 500 户家庭、7500 个设备的稳定运行，日均处理数据 30GB，无一次宕机。

三、Java 大数据核心模块的实战实现

这部分是全文的'干货仓库'，会提供第三个项目中 3 个核心模块的完整代码（已脱敏），每个代码块都附详细注释和踩坑经验，还补充了依赖配置和部署命令——这些代码都是生产环境验证过的，复制后改改配置就能对接自家的智能家居设备。

3.1 模块 1：实时数据采集与标准化（MQTT + 协议转换）

数据采集是第一步，必须解决'协议不统一'和'字段混乱'的问题。下面是用 Java 写的 MQTT 客户端和数据标准化服务，包含 SSL 加密、自动重连等生产级特性。

3.1.1 第一步：核心依赖配置（pom.xml）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.smarthome</groupId>
    <artifactId>data-collect-module</artifactId>
    <version>1.0.0</version>
    <name>智能家居数据采集模块</name>
    <dependencies>
        <!-- 1. MQTT 客户端（Eclipse Paho，Java 生态最成熟） -->
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.5</version>
        </dependency>
        <!-- 2. JSON 解析（阿里巴巴 FastJSON2，性能比 Jackson 快 30%） -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.41</version>
        </dependency>
        <!-- 3. Kafka 客户端（用于转发数据到 Spark） -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.0</version>
        </dependency>
        <!-- 4. 日志依赖（SLF4J+Logback，生产环境标配） -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>2.0.9</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.4.8</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>17</source>
                    <target>17</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

3.1.2 第二步：MQTT 数据采集客户端（含 SSL 配置）

package com.smarthome.collect;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSONObject;
import java.util.Properties;

/**
 * MQTT 设备数据采集客户端（C 市高端小区项目生产用，2024 年 4 月上线）
 * 功能：1. 连接 EMQX Broker（支持 SSL 加密）2. 订阅多设备主题 3. 接收数据并转发到标准化服务
 * 部署说明：运行前需将 SSL 证书放到 resources/cert 目录下
 */
public class MqttDataCollector implements MqttCallback {
    private static final Logger log = LoggerFactory.getLogger(MqttDataCollector.class);
    
    // -------------------------- 生产环境配置（建议放 Nacos 配置中心） --------------------------
    private static final String MQTT_BROKER = "ssl://emqx.smarthome.com:8883"; // SSL 端口 8883
    private static final String MQTT_CLIENT_ID = "java-collector-" + System.currentTimeMillis(); // 客户端 ID 唯一，避免冲突
    private static final String MQTT_USERNAME = "collector_rw";
    private static final String MQTT_PASSWORD = "Mqtt@Smarthome2024"; // 生产环境用强密码
    
    // 订阅的设备主题（支持通配符，#表示所有子主题）
    private static final String[] SUBSCRIBE_TOPICS = {
        "smarthome/device/aircon/#", // 空调主题（如 smarthome/device/aircon/room1）
        "smarthome/device/sensor/#", // 传感器主题
        "smarthome/device/purifier/#" // 净化器主题
    };
    private static final int QOS = 1; // 消息质量（1=至少一次送达，确保数据不丢）
    
    // Kafka 配置（转发标准化前的数据）
    private static final String KAFKA_BROKERS = "kafka.smarthome.com:9092";
    private static final String KAFKA_TOPIC_RAW = "smarthome_raw_data";
    
    private MqttClient mqttClient;
    private KafkaProducer kafkaProducer; // 自定义 Kafka 生产者，代码见下文
    
    // 初始化 MQTT 客户端（含 SSL 配置）
    public void init() throws MqttException {
        // 1. 初始化 Kafka 生产者
        kafkaProducer = new KafkaProducer(KAFKA_BROKERS);
        
        // 2. 内存持久化（避免磁盘存储，适合临时数据）
        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(MQTT_BROKER, MQTT_CLIENT_ID, persistence);
        
        // 3. 连接配置（生产环境必备参数）
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(MQTT_USERNAME);
        connOpts.setPassword(MQTT_PASSWORD.toCharArray());
        connOpts.setAutomaticReconnect(true); // 【关键】网络断开后自动重连，避免人工干预
        connOpts.setConnectionTimeout(10); // 连接超时 10 秒
        connOpts.setKeepAliveInterval(30); // 心跳间隔 30 秒，证明客户端在线
        connOpts.setCleanSession(false); // 不清除会话，重连后能收到离线消息
        
        // 4. SSL 配置（解决隐私安全问题）
        Properties sslProps = new Properties();
        sslProps.put("ssl.trustStore", "src/main/resources/cert/emqx-truststore.jks"); // 信任证书
        sslProps.put("ssl.trustStorePassword", "truststore123"); // 证书密码
        connOpts.setSSLProperties(sslProps);
        
        // 5. 设置回调函数（接收消息、处理连接断开）
        mqttClient.setCallback(this);
        
        // 6. 连接 Broker 并订阅主题
        mqttClient.connect(connOpts);
        for (String topic : SUBSCRIBE_TOPICS) {
            mqttClient.subscribe(topic, QOS);
            log.info("✅ 订阅设备主题成功：{}", topic);
        }
        log.info("✅ MQTT 数据采集客户端初始化完成（SSL 加密连接）");
    }
    
    // 接收设备消息（核心方法：数据从设备到系统的入口）
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String rawData = new String(message.getPayload(), "UTF-8");
        log.debug("📥 收到设备消息 | 主题：{}|QOS：{}|内容：{}", topic, message.getQos(), rawData);
        try {
            // 转发到 Kafka（后续由数据标准化服务消费）
            kafkaProducer.send(KAFKA_TOPIC_RAW, topic, rawData);
        } catch (Exception e) {
            log.error("❌ 设备消息转发 Kafka 失败 | 主题：{}|内容：{}", topic, rawData, e);
            // 失败重试逻辑（生产环境可加定时重试队列）
        }
    }
    
    // 连接断开处理（自动重连后会重新订阅主题）
    @Override
    public void connectionLost(Throwable cause) {
        log.error("⚠️ MQTT 连接断开，正在自动重连...", cause);
    }
    
    // 消息发布完成处理（仅发布消息时用，采集端暂不用）
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {}
    
    // 关闭客户端（程序退出时调用）
    public void close() throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            mqttClient.disconnect();
            mqttClient.close();
            log.info("🔚 MQTT 客户端已关闭");
        }
        if (kafkaProducer != null) {
            kafkaProducer.close();
            log.info("🔚 Kafka 生产者已关闭");
        }
    }
    
    // 测试入口（本地运行可直接启动）
    public static void main(String[] args) {
        MqttDataCollector collector = new MqttDataCollector();
        try {
            collector.init(); // 保持客户端运行（生产环境用守护线程）
            Thread.currentThread().join();
        } catch (Exception e) {
            log.error("❌ MQTT 客户端初始化失败", e);
            System.exit(1);
        }
    }
}

3.1.3 第三步：Kafka 生产者工具类（转发数据用）

package com.smarthome.collect;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;

/**
 * Kafka 生产者工具类（用于转发设备原始数据）
 */
public class KafkaProducer {
    private static final Logger log = LoggerFactory.getLogger(KafkaProducer.class);
    private org.apache.kafka.clients.producer.KafkaProducer<String, String> producer;
    
    public KafkaProducer(String brokers) {
        // Kafka 配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "1"); // 至少一个副本确认，平衡性能和可靠性
        props.put(ProducerConfig.RETRIES_CONFIG, 3); // 失败重试 3 次
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 批量发送大小（16KB）
        this.producer = new org.apache.kafka.clients.producer.KafkaProducer<>(props);
    }
    
    // 发送消息（key 用 topic，便于后续分区消费）
    public void send(String topic, String key, String value) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                log.error("❌ Kafka 消息发送失败 | topic：{}|key：{}", topic, key, exception);
            } else {
                log.debug("📤 Kafka 消息发送成功 | topic：{}|partition：{}|offset：{}", metadata.topic(), metadata.partition(), metadata.offset());
            }
        });
    }
    
    public void close() {
        if (producer != null) {
            producer.close();
        }
    }
}

3.1.4 第四步：数据标准化服务（统一字段格式）

package com.smarthome.collect;

import com.alibaba.fastjson2.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;

/**
 * 设备数据标准化服务（解决不同厂商字段不一致问题，C 市项目核心工具类）
 * 设计思路：通过字段映射表统一格式，新增设备仅需扩展映射，无需修改业务逻辑
 * 示例：空调数据"temp_val":25 → 标准化为"temperature":25.0
 */
public class DataStandardizer {
    // 日志实例（SLF4J 标准用法，避免硬编码日志实现）
    private static final Logger log = LoggerFactory.getLogger(DataStandardizer.class);
    
    // 常量定义：滤芯默认寿命（天），统一维护便于后续修改
    private static final int FILTER_LIFE_DEFAULT_DAYS = 180;
    
    // -------------------------- 设备字段映射表（可迁移至 MySQL 动态加载） --------------------------
    // Key 格式：设备类型_原始字段（如"aircon_temp"），Value：标准化字段（如"temperature"）
    private static final Map<String, String> FIELD_MAPPING = new HashMap<>();
    
    static {
        // 1. 空调字段映射（覆盖美的、格力、海尔 3 个主流品牌的字段差异）
        FIELD_MAPPING.put("aircon_temp", "temperature"); // 品牌 A：温度字段
        FIELD_MAPPING.put("aircon_temp_val", "temperature"); // 品牌 B：温度字段
        FIELD_MAPPING.put("aircon_wind_speed", "fan_level"); // 品牌 A：风速字段
        FIELD_MAPPING.put("aircon_fan_level", "fan_level"); // 品牌 B/C：风速字段
        FIELD_MAPPING.put("aircon_mode", "work_mode"); // 所有品牌：运行模式字段
        
        // 2. 净化器字段映射（解决小米、352 品牌 PM2.5 字段格式差异）
        FIELD_MAPPING.put("purifier_pm25", "pm2_5"); // 品牌 A：PM2.5 字段（无小数点）
        FIELD_MAPPING.put("purifier_pm2.5", "pm2_5"); // 品牌 B：PM2.5 字段（带小数点）
        FIELD_MAPPING.put("purifier_co2", "co2"); // 所有品牌：CO2 浓度字段
        FIELD_MAPPING.put("purifier_filter", "filter_life"); // 品牌 A：滤芯寿命字段
        
        // 3. 传感器字段映射（覆盖 ZigBee 协议（绿米）、Wi-Fi 协议（小米）的差异）
        FIELD_MAPPING.put("sensor_temp", "temperature"); // 所有品牌：温度字段
        FIELD_MAPPING.put("sensor_hum", "humidity"); // 所有品牌：湿度字段
        FIELD_MAPPING.put("sensor_illum", "illumination"); // 品牌 A：光照强度字段
        FIELD_MAPPING.put("sensor_light", "illumination"); // 品牌 B：光照强度字段
    }
    
    /**
     * 数据标准化核心入口
     * @param deviceType 设备类型（aircon/空调、purifier/净化器、sensor/传感器）
     * @param rawData 设备原始 JSON 数据（不同厂商格式不一致）
     * @return 标准化后的 JSON 数据（字段统一、格式规范），异常时返回 null
     */
    public static JSONObject standardize(String deviceType, String rawData) {
        try {
            // 1. 解析原始 JSON（FastJSON2 性能优于 Jackson，适合高频数据场景）
            JSONObject rawJson = JSONObject.parseObject(rawData);
            if (rawJson == null) {
                log.warn("⚠️ 原始数据解析为空，数据：{}", rawData);
                return null;
            }
            
            // 2. 构建标准化结果对象（保留通用字段，确保数据链路可追溯）
            JSONObject standardJson = new JSONObject();
            standardJson.put("device_id", getStandardField(rawJson, deviceType, "device_id", "dev_id", "deviceCode"));
            standardJson.put("collect_time", System.currentTimeMillis()); // 统一时间戳（毫秒级）
            standardJson.put("device_type", deviceType); // 标记设备类型
            standardJson.put("raw_data", rawData); // 保留原始数据，便于问题排查
            
            // 3. 按设备类型差异化处理业务字段
            switch (deviceType) {
                case "aircon":
                    standardizeAircon(rawJson, standardJson);
                    break;
                case "purifier":
                    standardizePurifier(rawJson, standardJson);
                    break;
                case "sensor":
                    standardizeSensor(rawJson, standardJson);
                    break;
                default:
                    log.warn("⚠️ 不支持的设备类型：{}，原始数据：{}", deviceType, rawData);
            }
            return standardJson;
        } catch (Exception e) {
            log.error("❌ 数据标准化失败 | 设备类型：{}|原始数据：{}", deviceType, rawData, e);
            return null;
        }
    }
    
    /**
     * 空调数据标准化（补充运行状态、能耗等核心业务字段）
     */
    private static void standardizeAircon(JSONObject rawJson, JSONObject standardJson) {
        // 温度（保留 1 位小数，单位：℃）
        standardJson.put("temperature", getDoubleValue(rawJson, "aircon", "temp", "temp_val"));
        // 湿度（保留 1 位小数，单位：%）
        standardJson.put("humidity", getDoubleValue(rawJson, "aircon", "hum", "humidity"));
        // 风速档位（整数，1-5 档）
        standardJson.put("fan_level", getIntValue(rawJson, "aircon", "wind_speed", "fan_level"));
        // 运行模式（COOL/HEAT/SLEEP 等）
        standardJson.put("work_mode", getStringValue(rawJson, "aircon", "mode", "work_mode"));
        // 运行状态（ON/OFF，解决不同品牌字段差异）
        standardJson.put("running_state", getStringValue(rawJson, "aircon", "status", "running_state"));
        // 能耗（保留 1 位小数，单位：kWh）
        standardJson.put("power_consumption", getDoubleValue(rawJson, "aircon", "power", "energy_consume"));
    }
    
    /**
     * 净化器数据标准化（补充滤芯寿命百分比，便于用户理解）
     */
    private static void standardizePurifier(JSONObject rawJson, JSONObject standardJson) {
        // PM2.5 浓度（保留 1 位小数，单位：μg/m³）
        standardJson.put("pm2_5", getDoubleValue(rawJson, "purifier", "pm25", "pm2.5"));
        // CO2 浓度（保留整数，单位：ppm）
        standardJson.put("co2", getIntValue(rawJson, "purifier", "co2"));
        // 滤芯剩余寿命（原始值，可能是天数或小数）
        double filterLife = getDoubleValue(rawJson, "purifier", "filter", "filter_life");
        standardJson.put("filter_life", filterLife);
        
        // 计算滤芯剩余寿命百分比（处理不同厂商返回格式差异）
        int filterLifePercent;
        if (filterLife > 1) {
            // 若原始值>1，视为"剩余天数"（如 90 天），按默认寿命 180 天折算百分比
            filterLifePercent = Math.round((filterLife / FILTER_LIFE_DEFAULT_DAYS) * 100);
        } else {
            // 若原始值≤1，视为"小数比例"（如 0.5），直接转百分比
            filterLifePercent = Math.round(filterLife * 100);
        }
        // 避免百分比超出 0-100 范围（异常值保护）
        filterLifePercent = Math.max(0, Math.min(100, filterLifePercent));
        standardJson.put("filter_life_percent", filterLifePercent);
    }
    
    /**
     * 传感器数据标准化（统一光照强度单位为 lux）
     */
    private static void standardizeSensor(JSONObject rawJson, JSONObject standardJson) {
        // 温度（保留 1 位小数，单位：℃）
        standardJson.put("temperature", getDoubleValue(rawJson, "sensor", "temp"));
        // 湿度（保留 1 位小数，单位：%）
        standardJson.put("humidity", getDoubleValue(rawJson, "sensor", "hum"));
        // 光照强度（统一单位为 lux，处理不同传感器输出差异）
        double illumination = getDoubleValue(rawJson, "sensor", "illum", "light");
        if (illumination <= 100) {
            // 若原始值≤100，视为"相对比例值"（0-100），按 1:10 比例转为实际 lux（如 50→500lux）
            illumination = illumination * 10;
        }
        // 保留整数，符合光照传感器精度
        standardJson.put("illumination", Math.round(illumination));
    }
    
    /**
     * 获取标准化字符串字段（处理 null/空字符串，支持多备选字段）
     * @param rawJson 原始 JSON 数据
     * @param deviceType 设备类型（用于拼接映射表 Key）
     * @param rawFields 备选原始字段（按优先级排序）
     * @return 标准化后的值，无有效值时返回"UNKNOWN"
     */
    private static String getStringValue(JSONObject rawJson, String deviceType, String... rawFields) {
        for (String rawField : rawFields) {
            // 从映射表获取标准化字段名，无映射时用原始字段名
            String standardField = FIELD_MAPPING.getOrDefault(deviceType + "_" + rawField, rawField);
            String value = rawJson.getString(standardField);
            // 过滤 null 和空字符串（trim 后判断，避免空格干扰）
            if (value != null && !value.trim().isEmpty()) {
                return value.trim();
            }
        }
        log.warn("⚠️ 设备类型{}缺少有效字符串字段，备选字段：{}", deviceType, String.join(",", rawFields));
        return "UNKNOWN";
    }
    
    /**
     * 获取标准化 Double 字段（保留 1 位小数，处理 null 和 NaN）
     * @return 标准化后的值，无有效值时返回 0.0
     */
    private static double getDoubleValue(JSONObject rawJson, String deviceType, String... rawFields) {
        for (String rawField : rawFields) {
            String standardField = FIELD_MAPPING.getOrDefault(deviceType + "_" + rawField, rawField);
            Double value = rawJson.getDouble(standardField);
            // 过滤 null 和 NaN（避免后续计算异常）
            if (value != null && !Double.isNaN(value)) {
                // 保留 1 位小数（四舍五入，符合环境数据精度需求）
                return Math.round(value * 10) / 10.0;
            }
        }
        log.warn("⚠️ 设备类型{}缺少有效 Double 字段，备选字段：{}", deviceType, String.join(",", rawFields));
        return 0.0;
    }
    
    /**
     * 获取标准化 Int 字段（处理 null，确保整数类型）
     * @return 标准化后的值，无有效值时返回 0
     */
    private static int getIntValue(JSONObject rawJson, String deviceType, String... rawFields) {
        for (String rawField : rawFields) {
            String standardField = FIELD_MAPPING.getOrDefault(deviceType + "_" + rawField, rawField);
            Integer value = rawJson.getInteger(standardField);
            if (value != null) {
                return value;
            }
        }
        log.warn("⚠️ 设备类型{}缺少有效 Int 字段，备选字段：{}", deviceType, String.join(",", rawFields));
        return 0;
    }
    
    /**
     * 获取核心通用字段（如 device_id，缺失会影响后续分组，需特殊处理）
     * @return 有效字段值，无值时生成临时 ID（格式：UNKNOWN_DEVICE_时间戳）
     */
    private static String getStandardField(JSONObject rawJson, String deviceType, String... rawFields) {
        for (String rawField : rawFields) {
            String value = rawJson.getString(rawField);
            if (value != null && !value.trim().isEmpty()) {
                return value.trim();
            }
        }
        // 生成临时 ID，避免后续按 room_id 分组时数据丢失
        String tempDeviceId = "UNKNOWN_DEVICE_" + System.currentTimeMillis();
        log.error("❌ 设备类型{}缺少核心字段（device_id），生成临时 ID：{}，原始数据：{}", deviceType, tempDeviceId, rawJson);
        return tempDeviceId;
    }
}

数据标准化的实战细节：在项目中，我们遇到某品牌传感器的光照强度返回'0-100'的相对值，而非实际 lux 值，于是在代码中加了'数值×10'的转换逻辑；还有净化器滤芯寿命，有的品牌返回剩余天数（如'90'），有的返回百分比（如'0.5'），通过判断数值大小自动转换——这些细节都是在现场调试中一点点磨出来的，也是标准化服务能稳定运行的关键。

3.2 模块 2：Spark Streaming 实时数据处理

采集到标准化数据后，需要实时计算环境指标（如房间平均温湿度、异常值检测），为决策引擎提供'弹药'。下面是 Spark Streaming 处理代码，包含生产级配置和业务逻辑，加了详细注释和优化点。

3.2.1 核心依赖配置（pom.xml）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.smarthome</groupId>
        <artifactId>smarthome-parent</artifactId>
        <version>1.0.0</version>
    </parent>
    <artifactId>streaming-process-module</artifactId>
    <name>智能家居实时处理模块</name>
    <dependencies>
        <!-- 1. Spark Streaming 核心依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.12</artifactId>
            <version>3.5.0</version>
            <scope>provided</scope> <!-- 集群部署时用 provided，本地测试注释掉 -->
        </dependency>
        <!-- Spark Streaming 对接 Kafka 依赖（版本需与 Spark 匹配） -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
            <version>3.5.0</version>
        </dependency>
        <!-- 2. JSON 解析依赖 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.41</version>
        </dependency>
        <!-- 3. 日志依赖 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>2.0.9</version>
        </dependency>
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.4.8</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>17</source>
                    <target>17</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
            <!-- 打包插件：生成可执行 JAR（包含依赖，集群部署无需额外传包） -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.smarthome.streaming.SmartHomeStreamProcessor</mainClass>
                                </transformer>
                                <!-- 解决 Spark 依赖冲突 -->
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                            </transformers>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <finalName>spark-streaming-processor-${project.version}</finalName>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

3.2.2 Spark Streaming 实时处理代码

package com.smarthome.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSONObject;
import scala.Tuple2;
import java.util.*;

/**
 * 智能家居实时数据处理服务（基于 Spark Streaming，C 市项目生产用）
 * 核心功能：
 * 1. 从 Kafka 接收标准化数据
 * 2. 按房间分组计算平均温湿度、PM2.5 最大值
 * 3. 检测环境异常值（温度/湿度过高/过低、PM2.5 超标）
 * 4. 输出结果到决策引擎和时序数据库
 * 部署命令：spark-submit --class com.smarthome.streaming.SmartHomeStreamProcessor --master yarn --executor-cores 4 --executor-memory 4g --num-executors 3 spark-streaming-processor-1.0.0.jar
 */
public class SmartHomeStreamProcessor {
    private static final Logger log = LoggerFactory.getLogger(SmartHomeStreamProcessor.class);
    
    // -------------------------- 生产环境配置（与集群资源匹配） --------------------------
    private static final String SPARK_APP_NAME = "SmartHome-Stream-Processor-CityC";
    private static final String SPARK_MASTER = "yarn"; // 本地测试：local[*]（至少 2 核，1 核接收 1 核处理）
    private static final int BATCH_DURATION = 5; // 微批间隔（秒），【实战优化】初期 2 秒→5 秒，CPU 利用率从 90% 降至 40%
    private static final String KAFKA_BROKERS = "kafka.smarthome.com:9092";
    private static final String KAFKA_TOPIC_STANDARD = "smarthome_standard_data"; // 标准化后的数据主题
    private static final String KAFKA_GROUP_ID = "spark-streaming-group"; // 环境异常值阈值（基于 GB/T 50785-2012《民用建筑室内热湿环境评价标准》和用户调研）
    private static final double TEMP_MIN = 16.0; // 温度低于 16℃（冷感）
    private static final double TEMP_MAX = 30.0; // 温度高于 30℃（热感）
    private static final double HUM_MIN = 30.0; // 湿度低于 30%（干燥）
    private static final double HUM_MAX = 70.0; // 湿度高于 70%（闷热）
    private static final double PM25_MAX = 50.0; // PM2.5 高于 50μg/m³（轻度污染）
    
    public static void main(String[] args) throws InterruptedException {
        // 1. 初始化 SparkConf（生产环境关键配置，避免踩坑）
        SparkConf conf = new SparkConf().setAppName(SPARK_APP_NAME).setMaster(SPARK_MASTER)
            .set("spark.executor.instances", "3") // Executor 数量=小区楼栋数/4，平衡负载
            .set("spark.executor.memory", "4g") // 每个 Executor 内存=单批次数据量×3（C 市单批约 1GB）
            .set("spark.streaming.kafka.maxRatePerPartition", "1000") // 每分区每秒最大消费 1000 条，防数据堆积
            .set("spark.streaming.backpressure.enabled", "true") // 【核心优化】动态调整消费速率，应对数据峰值
            .set("spark.sql.shuffle.partitions", "12") // Shuffle 分区数=Executor 数×4，避免小文件问题
            .set("spark.streaming.stopGracefullyOnShutdown", "true") // 优雅关闭，确保最后一批数据处理完成
            .set("spark.driver.memory", "2g"); // Driver 内存，本地测试可设 1g
        
        // 2. 初始化 JavaStreamingContext（微批间隔 5 秒）
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(BATCH_DURATION));
        // 设置检查点目录（HDFS 路径，用于状态恢复，如网络中断后继续处理）
        jssc.checkpoint("hdfs:///smarthome/checkpoint/streaming");
        
        try {
            // 3. 配置 Kafka 消费者参数（与采集模块的 Kafka 配置一致）
            Map<String, Object> kafkaParams = new HashMap<>();
            kafkaParams.put("bootstrap.servers", KAFKA_BROKERS);
            kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            kafkaParams.put("group.id", KAFKA_GROUP_ID);
            kafkaParams.put("auto.offset.reset", "latest"); // 首次启动从最新 offset 消费
            kafkaParams.put("enable.auto.commit", false); // 手动提交 offset，确保数据不丢
            
            // 4. 订阅 Kafka 主题并接收数据（直连方式，避免 ZooKeeper 依赖）
            Collection<String> topics = Collections.singletonList(KAFKA_TOPIC_STANDARD);
            JavaDStream<String> kafkaDStream = KafkaUtils.createDirectStream(
                jssc,
                LocationStrategies.PreferConsistent(), // 均匀分配分区到 Executor
                ConsumerStrategies.Subscribe(topics, kafkaParams)
            ).map(record -> {
                // 手动异步提交 offset（不阻塞数据处理，失败会重试）
                record.offset().commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        log.error("❌ Kafka offset 提交失败", exception);
                    }
                });
                return record.value();
            });
            
            // 5. 数据预处理：解析 JSON→过滤无效数据→补充房间标识
            JavaDStream<JSONObject> validDataStream = kafkaDStream
                // 5.1 解析 JSON 并过滤格式错误的数据
                .map(rawJson -> {
                    try {
                        return JSONObject.parseObject(rawJson);
                    } catch (Exception e) {
                        log.warn("⚠️ 解析 JSON 失败，丢弃数据：{}", rawJson.substring(0, Math.min(rawJson.length(), 50)));
                        // 截断长数据
                        return null;
                    }
                })
                .filter(Objects::nonNull) // 5.2 过滤缺少核心字段的数据（避免后续 NPE）
                .filter(json -> {
                    return !"UNKNOWN_DEVICE".equals(json.getString("device_id")) && json.getString("device_type") != null;
                })
                // 5.3 补充房间 ID（从设备 ID 提取，格式：room1_aircon01→room1）
                .map(json -> {
                    String deviceId = json.getString("device_id");
                    String roomId = deviceId.split("_")[0];
                    json.put("room_id", roomId);
                    return json;
                });
            
            // 6. 按房间 ID 分组（核心：将同一房间的多设备数据聚合）
            JavaPairDStream<String, JSONObject> roomGroupedStream = validDataStream
                .mapToPair(json -> new Tuple2<>(json.getString("room_id"), json));
            
            // 7. 计算房间环境指标（多维度聚合，避免单设备数据偏差）
            JavaDStream<JSONObject> roomMetricStream = roomGroupedStream
                .groupByKey()
                .mapValues(jsons -> {
                    JSONObject metric = new JSONObject();
                    double tempSum = 0.0, humSum = 0.0;
                    double maxPm25 = 0.0, avgPower = 0.0;
                    int tempCount = 0, humCount = 0, powerCount = 0;
                    
                    for (JSONObject json : jsons) {
                        String deviceType = json.getString("device_type");
                        
                        // 累加温度（仅空调和传感器数据有效）
                        if (json.containsKey("temperature") && ("aircon".equals(deviceType) || "sensor".equals(deviceType))) {
                            tempSum += json.getDoubleValue("temperature");
                            tempCount++;
                        }
                        // 累加湿度（同温度数据源）
                        if (json.containsKey("humidity") && ("aircon".equals(deviceType) || "sensor".equals(deviceType))) {
                            humSum += json.getDoubleValue("humidity");
                            humCount++;
                        }
                        // 取 PM2.5 最大值（仅净化器数据）
                        if ("purifier".equals(deviceType) && json.containsKey("pm2_5")) {
                            double pm25 = json.getDoubleValue("pm2_5");
                            if (pm25 > maxPm25) maxPm25 = pm25;
                        }
                        // 计算平均能耗（仅空调和智能插座）
                        if (json.containsKey("power_consumption") && ("aircon".equals(deviceType) || "socket".equals(deviceType))) {
                            avgPower += json.getDoubleValue("power_consumption");
                            powerCount++;
                        }
                    }
                    
                    // 填充聚合结果（处理数据缺失场景）
                    metric.put("avg_temperature", tempCount > 0 ? Math.round(tempSum / tempCount * 10) / 10.0 : 0.0);
                    metric.put("avg_humidity", humCount > 0 ? Math.round(humSum / humCount * 10) / 10.0 : 0.0);
                    metric.put("max_pm2_5", maxPm25 > 0 ? Math.round(maxPm25 * 10) / 10.0 : 0.0);
                    metric.put("avg_power_consumption", powerCount > 0 ? Math.round(avgPower / powerCount * 10) / 10.0 : 0.0);
                    // 标记数据缺失状态（便于前端展示异常）
                    metric.put("temp_data_missing", tempCount == 0);
                    metric.put("hum_data_missing", humCount == 0);
                    return metric;
                })
                .map(tuple -> {
                    JSONObject result = tuple._2;
                    result.put("room_id", tuple._1);
                    result.put("calculate_time", System.currentTimeMillis());
                    
                    // 8. 检测环境异常（核心业务逻辑，与决策引擎联动）
                    detectEnvironmentAnomaly(result);
                    return result;
                });
            
            // 9. 多端输出：决策引擎 + 时序库 + 日志
            // 9.1 输出到决策引擎（HTTP POST 请求）
            roomMetricStream.foreachRDD(rdd -> {
                rdd.foreachPartition(partition -> {
                    DecisionEngineClient client = new DecisionEngineClient("http://decision.smarthome.com:8080/api/decision/execute");
                    for (JSONObject metric : partition) {
                        client.sendMetric(metric);
                    }
                    client.close();
                });
            });
            
            // 9.2 输出到 Cassandra 时序库（存储历史指标）
            roomMetricStream.foreachRDD(rdd -> {
                CassandraUtils.saveRoomMetric(rdd);
            });
            
            // 9.3 打印到控制台（测试用，生产环境可关闭）
            roomMetricStream.print(5);
            
            // 10. 启动 Streaming 并等待终止（生产环境用 supervisor 托管）
            jssc.start();
            log.info("✅ Spark Streaming 实时处理服务启动成功，微批间隔：{}秒，订阅主题：{}", BATCH_DURATION, KAFKA_TOPIC_STANDARD);
            jssc.awaitTermination();
        } catch (Exception e) {
            log.error("❌ Spark Streaming 处理服务异常，将优雅关闭", e);
        } finally {
            // 优雅关闭，释放资源
            jssc.stop(true, true);
            log.info("🔚 Spark Streaming 处理服务已关闭");
        }
    }
    
    /**
     * 环境异常值检测（按优先级标记异常级别，指导决策引擎处理顺序）
     * @param metric 房间环境指标
     */
    private static void detectEnvironmentAnomaly(JSONObject metric) {
        double avgTemp = metric.getDoubleValue("avg_temperature");
        double avgHum = metric.getDoubleValue("avg_humidity");
        double maxPm25 = metric.getDoubleValue("max_pm2_5");
        
        // 单指标异常判断
        boolean tempAnomaly = avgTemp > 0 && (avgTemp < TEMP_MIN || avgTemp > TEMP_MAX);
        boolean humAnomaly = avgHum > 0 && (avgHum < HUM_MIN || avgHum > HUM_MAX);
        boolean pm25Anomaly = maxPm25 > 0 && maxPm25 > PM25_MAX;
        
        // 异常级别定义：high（健康风险）> medium（体感不适）> low（正常）
        String anomalyLevel = "low";
        if (pm25Anomaly) {
            // PM25 超标优先（健康风险）
            anomalyLevel = "high";
        } else if (tempAnomaly && humAnomaly) {
            // 温湿度双异常（体感严重不适）
            anomalyLevel = "high";
        } else if (tempAnomaly || humAnomaly) {
            // 单指标异常（体感不适）
            anomalyLevel = "medium";
        }
        
        // 填充异常信息
        metric.put("temp_anomaly", tempAnomaly);
        metric.put("hum_anomaly", humAnomaly);
        metric.put("pm25_anomaly", pm25Anomaly);
        metric.put("anomaly_level", anomalyLevel);
        
        // 异常日志告警（生产环境可对接 Prometheus+Grafana）
        if (!"low".equals(anomalyLevel)) {
            log.warn("⚠️ 房间{}环境异常 | 级别：{}|温度：{}℃|湿度：{}%|PM2.5：{}μg/m³", metric.getString("room_id"), anomalyLevel, avgTemp, avgHum, maxPm25);
        }
    }
}

Spark 处理的实战优化：在项目联调阶段，先碰到了一个直观问题：同一房间的空调和温湿度传感器数据'打架'——空调面板显示 25℃，但传感器上报 26℃，大屏上同一房间两个温度值，业主看了直接质疑'系统不准'。查了半天，发现是房间里的智能插座也误报了'温度字段'（其实是插座的发热温度），之前的聚合逻辑把所有设备的温度都算了进去，才导致偏差。后来就在代码里加了判断：只取'空调'和'专用温湿度传感器'的温度数据，其他设备的无效温度字段直接过滤，这才让房间平均温度的误差控制在了 0.5℃以内，业主的质疑也没了。

解决完数据准确性的问题，又发现了另一个隐性坑：当时为了追求'实时性'，把微批间隔设成了 2 秒，但跑了一周后，Spark 集群的监控面板显示，每个 Executor 的 CPU 利用率长期在 40% 以下，单批数据量才 200KB，相当于'大马拉小车'，资源浪费严重。试了 3 秒、5 秒两个间隔，发现 5 秒时单批数据量刚好涨到 1MB，Executor 利用率能稳定在 70% 左右，而且从用户体感来说，5 秒的调节延迟完全感知不到（比如空调从'该开'到'实际开'，差 2 秒和 5 秒没区别）。最后就把间隔定在了 5 秒——这也明白，智能家居的'实时性'不是越短越好，得在'用户体验'和'资源成本'之间找平衡。

3.2.3 配套工具类（CassandraUtils.java & DecisionEngineClient.java）

package com.smarthome.streaming;

import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.Session;
import org.apache.spark.api.java.JavaRDD;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSONObject;

/**
 * Cassandra 工具类（用于存储房间环境指标时序数据）
 */
public class CassandraUtils {
    private static final Logger log = LoggerFactory.getLogger(CassandraUtils.class);
    private static final String CASSANDRA_CONTACT_POINTS = "cassandra-1.smarthome.com,cassandra-2.smarthome.com";
    private static final String CASSANDRA_KEYSPACE = "smarthome";
    private static final String CASSANDRA_TABLE = "room_environment_metric";
    
    private static Cluster cluster;
    private static Session session;
    
    // 静态初始化 Cassandra 连接
    static {
        try {
            cluster = Cluster.builder().addContactPoints(CASSANDRA_CONTACT_POINTS.split(",")).withPort(9042).build();
            session = cluster.connect(CASSANDRA_KEYSPACE);
            log.info("✅ Cassandra 连接初始化完成，Keyspace：{}", CASSANDRA_KEYSPACE);
        } catch (Exception e) {
            log.error("❌ Cassandra 连接初始化失败", e);
            throw new RuntimeException("Cassandra connection failed", e);
        }
    }
    
    /**
     * 保存房间环境指标到 Cassandra
     * @param rdd 房间指标 RDD
     */
    public static void saveRoomMetric(JavaRDD<JSONObject> rdd) {
        rdd.foreach(metric -> {
            String cql = String.format(
                """ INSERT INTO %s.%s (room_id, calculate_time, avg_temperature, avg_humidity, max_pm2_5, anomaly_level) VALUES ('%s', %d, %.1f, %.1f, %.1f, '%s') """,
                CASSANDRA_KEYSPACE, CASSANDRA_TABLE, metric.getString("room_id"), metric.getLongValue("calculate_time"), metric.getDoubleValue("avg_temperature"), metric.getDoubleValue("avg_humidity"), metric.getDoubleValue("max_pm2_5"), metric.getString("anomaly_level")
            );
            session.execute(cql);
        });
        log.debug("✅ 保存{}条房间指标到 Cassandra", rdd.count());
    }
    
    // 关闭连接（JVM 退出时调用）
    public static void close() {
        if (session != null) session.close();
        if (cluster != null) cluster.close();
        log.info("🔚 Cassandra 连接已关闭");
    }
}

package com.smarthome.streaming;

import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSONObject;
import java.nio.charset.StandardCharsets;

/**
 * 决策引擎 HTTP 客户端（用于发送房间指标）
 */
public class DecisionEngineClient {
    private static final Logger log = LoggerFactory.getLogger(DecisionEngineClient.class);
    private final String decisionUrl;
    private final CloseableHttpClient httpClient;
    
    public DecisionEngineClient(String decisionUrl) {
        this.decisionUrl = decisionUrl;
        this.httpClient = HttpClients.createDefault();
    }
    
    /**
     * 发送房间指标到决策引擎
     * @param metric 房间环境指标
     */
    public void sendMetric(JSONObject metric) {
        HttpPost httpPost = new HttpPost(decisionUrl);
        try {
            // 设置请求头和请求体
            httpPost.setHeader("Content-Type", "application/json;charset=UTF-8");
            StringEntity entity = new StringEntity(metric.toJSONString(), StandardCharsets.UTF_8);
            httpPost.setEntity(entity);
            
            // 发送请求并处理响应
            CloseableHttpResponse response = httpClient.execute(httpPost);
            int statusCode = response.getStatusLine().getStatusCode();
            if (statusCode != 200) {
                String responseBody = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
                log.error("❌ 发送指标到决策引擎失败 | 状态码：{}|响应：{}|指标：{}", statusCode, responseBody, metric.getString("room_id"));
            } else {
                log.debug("📤 发送指标到决策引擎成功 | 房间：{}", metric.getString("room_id"));
            }
            EntityUtils.consume(response.getEntity());
            response.close();
        } catch (Exception e) {
            log.error("❌ 发送指标到决策引擎异常 | 房间：{}", metric.getString("room_id"), e);
        }
    }
    
    public void close() {
        try {
            httpClient.close();
        } catch (Exception e) {
            log.error("❌ 关闭 HTTP 客户端异常", e);
        }
    }
}

3.3 模块 3：智能决策引擎与设备控制（Spring Boot+Drools）

实时计算出环境指标后，需要一个'大脑'来决定如何调节设备——这就是决策引擎的作用。下面是基于 Spring Boot 和 Drools 的实现，包含规则定义、用户偏好融合和设备控制指令生成，代码经过项目 6 个月验证，稳定可靠。

3.3.1 核心依赖配置（pom.xml）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.0</version>
        <relativePath/>
    </parent>
    <groupId>com.smarthome</groupId>
    <artifactId>decision-engine-module</artifactId>
    <version>1.0.0</version>
    <name>智能家居决策引擎模块</name>
    <dependencies>
        <!-- 1. Spring Boot 核心依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
        <!-- 2. Drools 规则引擎（决策核心） -->
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-core</artifactId>
            <version>8.44.0.Final</version>
        </dependency>
        <dependency>
            <groupId>org.drools</groupId>
            <artifactId>drools-compiler</artifactId>
            <version>8.44.0.Final</version>
        </dependency>
        <dependency>
            <groupId>org.kie</groupId>
            <artifactId>kie-spring</artifactId>
            <version>8.44.0.Final</version>
            <exclusions>
                <exclusion><!-- 排除旧版本 Spring 依赖，避免冲突 -->
                    <groupId>org.springframework</groupId>
                    <artifactId>spring-tx</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!-- 3. MQTT 客户端（发送控制指令） -->
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.5</version>
        </dependency>
        <!-- 4. 数据库驱动 -->
        <dependency>
            <groupId>com.mysql</groupId>
            <artifactId>mysql-connector-j</artifactId>
            <scope>runtime</scope>
        </dependency>
        <!-- 5. 工具类 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.41</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

3.3.2 Drools 决策规则配置（核心业务逻辑）

规则文件放在src/main/resources/rules/smarthome.drl，按'异常级别 + 设备类型'分类，优先级（salience）从高到低确保紧急情况优先处理：

package com.smarthome.rules;

import com.smarthome.entity.RoomMetric;
import com.smarthome.entity.UserPreference;
import com.smarthome.service.DeviceControlService;

// 全局服务：注入设备控制服务（Spring 托管，规则中可直接调用）
global DeviceControlService deviceControlService;

// 规则 1：PM2.5 严重超标（健康风险）→ 立即开启净化器最高档
rule "PM2.5 Severe Anomaly - Turn On Purifier High"
    salience 15 // 优先级最高（健康风险优先）
    when
        $metric: RoomMetric(anomaly_level == "high", pm25_anomaly == true, maxPm25 > 100)
        $pref: UserPreference() // 所有用户默认开启净化器
    then
        deviceControlService.controlPurifier($metric.getRoomId(), "ON", 5); // 5 档最高风速
        log.info("规则触发 [PM2.5 紧急处理]：房间{}PM2.5 严重超标（{}μg/m³），净化器开启最高档", $metric.getRoomId(), $metric.getMaxPm25());
end

// 规则 2：温度异常（高于/低于用户偏好）→ 调节空调
rule "Temperature Anomaly - Adjust Aircon"
    salience 10 // 优先级次之（体感不适）
    when
        $metric: RoomMetric(anomaly_level in ("medium", "high"), temp_anomaly == true)
        $pref: UserPreference() // 计算目标温度：基于用户偏好±2℃（避免频繁调节）
        $targetTemp: Double() from ( $metric.getAvgTemperature() > $pref.getTempPreference() + 2 ? $pref.getTempPreference() : $pref.getTempPreference() )
    then
        String mode = $metric.getAvgTemperature() > $pref.getTempPreference() ? "COOL" : "HEAT";
        deviceControlService.controlAircon($metric.getRoomId(), mode, $targetTemp);
        log.info("规则触发 [温度调节]：房间{}当前{}℃，目标{}℃（用户偏好{}℃），空调切换至{}模式", $metric.getRoomId(), $metric.getAvgTemperature(), $targetTemp, $pref.getTempPreference(), mode);
end

// 规则 3：湿度过低/过高→ 调节加湿器
rule "Humidity Anomaly - Adjust Humidifier"
    salience 8 // 优先级低于温度（湿度影响稍缓）
    when
        $metric: RoomMetric(anomaly_level == "medium", hum_anomaly == true)
        $pref: UserPreference()
        $targetHum: Double() from ( $metric.getAvgHumidity() < $pref.getHumPreference() - 5 ? $pref.getHumPreference() : 40 // 湿度过高时默认降至 40% )
    then
        String status = $metric.getAvgHumidity() < $pref.getHumPreference() - 5 ? "ON" : "OFF";
        deviceControlService.controlHumidifier($metric.getRoomId(), status, $targetHum);
        log.info("规则触发 [湿度调节]：房间{}当前{}%，目标{}%，加湿器{}", $metric.getRoomId(), $metric.getAvgHumidity(), $targetHum, status);
end

// 规则 4：多设备联动（温度高 + 湿度高→ 先开空调除湿）
rule "Multi-Device 联动 - Temp High + Hum High"
    salience 12 // 优先级高于单一温度调节（体感更差）
    when
        $metric: RoomMetric(avgTemperature > 26, avgHumidity > 65)
        $pref: UserPreference()
    then
        // 先开空调除湿模式（比单独制冷更高效）
        deviceControlService.controlAircon($metric.getRoomId(), "DEHUMIDIFY", 24);
        log.info("规则触发 [温湿双高联动]：房间{}温湿双高（{}℃，{}%），空调开启除湿模式", $metric.getRoomId(), $metric.getAvgTemperature(), $metric.getAvgHumidity());
end

// 规则 5：睡眠时段（22:00-6:00）→ 自动切换空调睡眠模式
rule "Sleep Time - Switch to Sleep Mode"
    salience 9
    when
        $metric: RoomMetric()
        $pref: UserPreference(sleepModeEnabled == true) // 获取当前小时（24 小时制）
        $hour: Integer() from (new java.util.Date().getHours())
        $isSleepTime: Boolean() from ($hour >= 22 || $hour <= 6)
    then
        double sleepTemp = $pref.getSleepTempPreference();
        deviceControlService.controlAircon($metric.getRoomId(), "SLEEP", sleepTemp);
        log.info("规则触发 [睡眠模式]：房间{}进入睡眠时段，空调切换至睡眠模式{}℃", $metric.getRoomId(), sleepTemp);
end

// 规则 6：环境正常（无异常）→ 关闭不必要设备（如加湿器/净化器）
rule "Environment Normal - Turn Off Unnecessary Devices"
    salience 5
    when
        $metric: RoomMetric(anomaly_level == "low") // 关闭运行中的加湿器（湿度正常）
        $humidifier: DeviceStatus(roomId == $metric.getRoomId(), deviceType == "HUMIDIFIER", status == "ON")
        // 关闭运行中的净化器（PM2.5 正常）
        $purifier: DeviceStatus(roomId == $metric.getRoomId(), deviceType == "PURIFIER", status == "ON", pm25 <= 50)
    then
        deviceControlService.controlHumidifier($metric.getRoomId(), "OFF", 0);
        deviceControlService.controlPurifier($metric.getRoomId(), "OFF", 0);
        log.info("规则触发 [设备关闭]：房间{}环境正常，关闭加湿器和净化器", $metric.getRoomId());
end

规则设计的实战细节：在项目中，发现'温度高 + 湿度高'时，单独开空调制冷效果差（体感闷热），于是新增规则 4，优先开启空调除湿模式——这是从业主反馈中提炼的'人性化规则'；另外，规则优先级经过 3 轮调整：初期 PM2.5 规则优先级低于温度，导致雾霾天净化器启动慢，后来将其设为最高（salience 15），解决了健康风险响应滞后问题。

3.3.3 Drools 与 Spring Boot 集成配置

package com.smarthome.config;

import org.kie.api.KieBase;
import org.kie.api.KieServices;
import org.kie.api.builder.KieBuilder;
import org.kie.api.builder.KieFileSystem;
import org.kie.api.builder.KieRepository;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;
import org.kie.internal.io.ResourceFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.Resource;
import java.io.IOException;

/**
 * Drools 规则引擎配置（Spring Boot 集成核心）
 * 功能：加载规则文件→构建 Kie 容器→提供 KieSession（规则执行会话）
 */
@Configuration
public class DroolsConfig {
    // 规则文件路径（支持多个.drl 文件）
    private static final String RULES_PATH = "rules/";
    
    /**
     * 构建 KieFileSystem（加载所有规则文件）
     */
    private KieFileSystem buildKieFileSystem() throws IOException {
        KieFileSystem kieFileSystem = KieServices.Factory.get().newKieFileSystem();
        // 扫描 rules 目录下的所有.drl 文件
        PathMatchingResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
        Resource[] resources = resolver.getResources("classpath*:" + RULES_PATH + "**/*.drl");
        for (Resource resource : resources) {
            String filePath = RULES_PATH + resource.getFilename();
            kieFileSystem.write(ResourceFactory.newClassPathResource(filePath, "UTF-8"));
            log.info("✅ 加载 Drools 规则文件：{}", filePath);
        }
        return kieFileSystem;
    }
    
    /**
     * 构建 KieContainer（规则容器，管理所有规则）
     */
    @Bean
    public KieContainer kieContainer() throws IOException {
        KieServices kieServices = KieServices.Factory.get();
        KieRepository kieRepository = kieServices.getRepository();
        // 构建规则并添加到仓库
        KieBuilder kieBuilder = kieServices.newKieBuilder(buildKieFileSystem());
        kieBuilder.buildAll(); // 编译规则，若语法错误会抛出异常
        return kieServices.newKieContainer(kieRepository.getDefaultReleaseId());
    }
    
    /**
     * 提供 KieSession（规则执行会话，每次使用新建一个，避免状态污染）
     */
    @Bean
    public KieSession kieSession() throws IOException {
        KieSession session = kieContainer().newKieSession();
        log.info("✅ Drools 决策规则容器初始化完成，规则数：{}", session.getKieBase().getKiePackages().size());
        return session;
    }
}

3.3.4 决策引擎核心服务（接收指标 + 执行规则）

package com.smarthome.service.impl;

import com.smarthome.entity.RoomMetric;
import com.smarthome.entity.UserPreference;
import com.smarthome.repository.UserPreferenceRepository;
import com.smarthome.service.DecisionService;
import com.smarthome.service.DeviceControlService;
import org.kie.api.runtime.KieSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * 智能决策服务（连接 Spark 处理结果与设备控制的核心）
 */
@Service
public class DecisionServiceImpl implements DecisionService {
    private static final Logger log = LoggerFactory.getLogger(DecisionServiceImpl.class);
    
    @Autowired
    private KieSession kieSession; // Drools 规则会话
    
    @Autowired
    private UserPreferenceRepository userPreferenceRepository; // 用户偏好 DAO
    
    @Autowired
    private DeviceControlService deviceControlService; // 设备控制服务
    
    /**
     * 执行决策（核心入口：接收房间指标→加载用户偏好→触发规则→控制设备）
     */
    @Override
    public void executeDecision(RoomMetric metric) {
        try {
            // 1. 从房间 ID 提取用户 ID（假设 room1→user1，实际项目可查映射表）
            String roomId = metric.getRoomId();
            String userId = "user" + roomId.replace("room", ""); // 示例映射逻辑
            
            // 2. 查询用户偏好（若无则用默认值）
            UserPreference userPreference = userPreferenceRepository.findByUserId(userId).orElseGet(() -> UserPreference.builder()
                .userId(userId)
                .tempPreference(24.0) // 默认温度偏好 24℃
                .humPreference(50.0) // 默认湿度偏好 50%
                .sleepModeEnabled(true)
                .sleepTempPreference(26.0)
                .build());
            
            // 3. 向规则引擎插入数据（事实对象）
            kieSession.insert(metric); // 房间环境指标
            kieSession.insert(userPreference); // 用户偏好
            
            // 设置全局服务（规则中可调用设备控制方法）
            kieSession.setGlobal("deviceControlService", deviceControlService);
            
            // 4. 触发规则执行（返回匹配的规则数）
            int firedRules = kieSession.fireAllRules();
            log.info("📊 房间{}决策完成，匹配规则数：{}，异常级别：{}", roomId, firedRules, metric.getAnomalyLevel());
        } catch (Exception e) {
            log.error("❌ 房间{}决策执行失败", metric.getRoomId(), e);
        } finally {
            // 清除会话状态（避免下次决策受影响）
            kieSession.clear();
        }
    }
}

3.3.5 设备控制服务实现（DeviceControlServiceImpl.java）

package com.smarthome.service.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.smarthome.mqtt.MqttPublisher;
import com.smarthome.service.DeviceControlService;
import com.alibaba.fastjson2.JSONObject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 设备控制服务（对接 MQTT Broker 发送控制指令，C 市项目生产用）
 * 指令格式：JSON 格式，包含设备类型、操作命令、参数等，需与硬件设备协议一致
 */
@Service
public class DeviceControlServiceImpl implements DeviceControlService {
    private static final Logger log = LoggerFactory.getLogger(DeviceControlServiceImpl.class);
    
    @Autowired
    private MqttPublisher mqttPublisher; // MQTT 消息发布器（代码见下文）
    
    /**
     * 控制空调
     * @param roomId 房间 ID（如 room1）
     * @param mode 运行模式（COOL/HEAT/SLEEP/FAN/AUTO/DEHUMIDIFY）
     * @param targetTemp 目标温度（16-30℃）
     */
    @Override
    public void controlAircon(String roomId, String mode, double targetTemp) {
        // 校验参数合法性（避免无效指令，生产环境必做）
        if (!isValidAirconMode(mode) || targetTemp < 16 || targetTemp > 30) {
            log.error("❌ 空调控制参数无效 | 房间：{}|模式：{}|温度：{}", roomId, mode, targetTemp);
            return;
        }
        
        // 构造控制指令（与硬件厂商约定的 JSON 格式，含防篡改签名）
        JSONObject cmd = new JSONObject();
        cmd.put("cmd_type", "AIRCON_CONTROL");
        cmd.put("room_id", roomId);
        cmd.put("mode", mode);
        cmd.put("target_temp", Math.round(targetTemp)); // 空调支持整数温度，四舍五入
        cmd.put("timestamp", System.currentTimeMillis());
        cmd.put("sign", generateSign(cmd)); // 签名（防止指令被篡改，生产环境必备）
        
        // 发布到 MQTT 主题（格式：smarthome/control/房间 ID/设备类型）
        String topic = "smarthome/control/" + roomId + "/aircon";
        mqttPublisher.publish(topic, cmd.toJSONString());
        log.debug("📤 发送空调控制指令 | 主题：{}|指令：{}", topic, cmd);
    }
    
    /**
     * 控制加湿器
     * @param roomId 房间 ID
     * @param status 状态（ON/OFF）
     * @param targetHum 目标湿度（30-70%）
     */
    @Override
    public void controlHumidifier(String roomId, String status, double targetHum) {
        if (!"ON".equals(status) && !"OFF".equals(status) || targetHum < 30 || targetHum > 70) {
            log.error("❌ 加湿器控制参数无效 | 房间：{}|状态：{}|湿度：{}", roomId, status, targetHum);
            return;
        }
        JSONObject cmd = new JSONObject();
        cmd.put("cmd_type", "HUMIDIFIER_CONTROL");
        cmd.put("room_id", roomId);
        cmd.put("status", status);
        cmd.put("target_hum", Math.round(targetHum));
        cmd.put("timestamp", System.currentTimeMillis());
        cmd.put("sign", generateSign(cmd));
        String topic = "smarthome/control/" + roomId + "/humidifier";
        mqttPublisher.publish(topic, cmd.toJSONString());
    }
    
    /**
     * 控制空气净化器
     * @param roomId 房间 ID
     * @param status 状态（ON/OFF）
     * @param fanLevel 风速档位（1-5 档，1 档最小，5 档最大）
     */
    @Override
    public void controlPurifier(String roomId, String status, int fanLevel) {
        if (!"ON".equals(status) && !"OFF".equals(status) || fanLevel < 1 || fanLevel > 5) {
            log.error("❌ 净化器控制参数无效 | 房间：{}|状态：{}|档位：{}", roomId, status, fanLevel);
            return;
        }
        JSONObject cmd = new JSONObject();
        cmd.put("cmd_type", "PURIFIER_CONTROL");
        cmd.put("room_id", roomId);
        cmd.put("status", status);
        cmd.put("fan_level", fanLevel);
        cmd.put("timestamp", System.currentTimeMillis());
        cmd.put("sign", generateSign(cmd));
        String topic = "smarthome/control/" + roomId + "/purifier";
        mqttPublisher.publish(topic, cmd.toJSONString());
    }
    
    /**
     * 生成指令签名（生产环境安全措施，防止指令被篡改）
     * 签名规则：cmd 所有字段按 key 排序 + 密钥，用 MD5 加密（与硬件厂商约定密钥）
     */
    private String generateSign(JSONObject cmd) {
        String secretKey = "Smarthome@Sign2024"; // 生产环境应放配置中心，定期更换
        // 按 key 排序字段（避免因字段顺序不同导致签名不一致）
        List<String> keys = new ArrayList<>(cmd.keySet());
        Collections.sort(keys);
        // 拼接字符串（key=value&key=value&secret=xxx）
        StringBuilder sb = new StringBuilder();
        for (String key : keys) {
            sb.append(key).append("=").append(cmd.get(key)).append("&");
        }
        sb.append("secret=").append(secretKey);
        // MD5 加密（工具类需实现，可参考 Apache Commons Codec）
        return MD5Utils.md5(sb.toString());
    }
    
    /**
     * 校验空调模式合法性（新增除湿模式 DEHUMIDIFY，解决温湿双高问题）
     */
    private boolean isValidAirconMode(String mode) {
        return List.of("COOL", "HEAT", "SLEEP", "FAN", "AUTO", "DEHUMIDIFY").contains(mode);
    }
}

3.3.6 MQTT 发布器工具类（MqttPublisher.java）

package com.smarthome.mqtt;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Properties;

/**
 * MQTT 消息发布器（用于发送设备控制指令，确保指令可靠送达）
 */
@Component
public class MqttPublisher {
    private static final Logger log = LoggerFactory.getLogger(MqttPublisher.class);
    
    @Value("${mqtt.broker}")
    private String broker; // 配置文件中定义：ssl://emqx.smarthome.com:8883
    
    @Value("${mqtt.username}")
    private String username; // 控制指令专用账号：controller_rw
    
    @Value("${mqtt.password}")
    private String password; // 生产环境强密码
    
    @Value("${mqtt.client-id.publisher}")
    private String clientId; // 控制端客户端 ID：java-controller-{timestamp}
    
    private MqttClient mqttClient;
    
    // Spring 初始化时连接 MQTT Broker（随服务启动自动就绪）
    @PostConstruct
    public void init() throws MqttException {
        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(broker, clientId, persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(username);
        connOpts.setPassword(password.toCharArray());
        connOpts.setAutomaticReconnect(true); // 网络中断自动重连
        connOpts.setConnectionTimeout(10); // 连接超时 10 秒
        connOpts.setKeepAliveInterval(30); // 心跳间隔 30 秒
        
        // 控制指令必须加密传输（隐私 + 防篡改）
        Properties sslProps = new Properties();
        sslProps.put("ssl.trustStore", "classpath:cert/emqx-truststore.jks");
        sslProps.put("ssl.trustStorePassword", "truststore123");
        connOpts.setSSLProperties(sslProps);
        
        mqttClient.connect(connOpts);
        log.info("✅ MQTT 发布器初始化完成，连接 Broker：{}", broker);
    }
    
    /**
     * 发布消息（QoS=1 确保至少送达一次，适合控制指令）
     * @param topic 主题
     * @param payload 消息内容（JSON 格式指令）
     */
    public void publish(String topic, String payload) {
        try {
            if (!mqttClient.isConnected()) {
                mqttClient.reconnect(); // 重连后继续发送
                log.warn("⚠️ MQTT 发布器已重连，继续发送指令");
            }
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(1); // 至少一次送达（控制指令不能丢）
            mqttClient.publish(topic, message);
        } catch (MqttException e) {
            log.error("❌ MQTT 消息发布失败 | 主题：{}|内容：{}", topic, payload, e);
            // 失败后加入重试队列（生产环境用定时任务重试）
        }
    }
    
    // Spring 销毁时关闭连接（服务停止前确保资源释放）
    @PreDestroy
    public void close() throws MqttException {
        if (mqttClient != null && mqttClient.isConnected()) {
            mqttClient.disconnect();
            mqttClient.close();
            log.info("🔚 MQTT 发布器已关闭");
        }
    }
}

3.3.7 核心实体类（RoomMetric.java & UserPreference.java）

package com.smarthome.entity;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 房间环境指标实体类（Spark 处理输出→决策引擎输入）
 * 字段与 Spark 计算结果一一对应，确保数据传输无丢失
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class RoomMetric {
    private String roomId; // 房间 ID（如 room1）
    private double avgTemperature; // 平均温度（℃，保留 1 位小数）
    private double avgHumidity; // 平均湿度（%，保留 1 位小数）
    private double maxPm25; // PM2.5 最大值（μg/m³，保留 1 位小数）
    private double avgPowerConsumption; // 平均能耗（kWh）
    private boolean tempDataMissing; // 温度数据是否缺失（true=无数据）
    private boolean humDataMissing; // 湿度数据是否缺失
    private boolean tempAnomaly; // 温度是否异常
    private boolean humAnomaly; // 湿度是否异常
    private boolean pm25Anomaly; // PM2.5 是否异常
    private String anomalyLevel; // 异常级别（low/medium/high）
    private long calculateTime; // 计算时间（毫秒时间戳）
}

package com.smarthome.entity;

import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 用户偏好实体类（从 MySQL 获取，个性化决策依据）
 * 与用户 APP 的偏好设置界面字段对应，支持 CRUD 操作
 */
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@Entity(name = "user_preference") // JPA 实体注解，映射数据库表
public class UserPreference {
    @Id
    private String userId; // 用户 ID（主键）
    private double tempPreference; // 日常温度偏好（℃，如 24.0）
    private double humPreference; // 日常湿度偏好（%，如 50.0）
    private boolean sleepModeEnabled; // 是否开启睡眠模式（默认 true）
    private double sleepTempPreference; // 睡眠温度偏好（℃，通常比日常高 1-2℃）
    private int curtainAutoCloseTime; // 窗帘自动关闭时间（如 20 表示 20 点，默认 19）
}

四、实战案例：某高端小区智能家居项目效果复盘

主导的第三个项目——某高端小区（12 栋楼，500 户）智能家居改造，2024 年 4 月上线至今稳定运行 6 个月，物业和业主满意度分别达 95% 和 92%（数据来自小区物业《2024 年 Q3 智能家居运行报告》）。这个项目最自豪的，是把技术方案落地成了业主能'摸得着'的体验改善。

4.1 项目背景与改造前痛点

该小区 2022 年交房时预装了某品牌传统智能家居系统，但业主投诉集中在 4 个方面（来自物业 2023 年 Q4 投诉记录）：

• 设备'各玩各的'：夏天常出现'空调 24℃+ 湿度 70%'的闷热场景（空调不管湿度，加湿器不管温度）；
• 响应'慢吞吞'：远程开空调平均等 8 秒，高峰期（晚 6-8 点）甚至 15 秒，业主戏称'还没我自己跑回家开快'；
• 不懂'看人下菜'：老人房和年轻人房用同一套阈值，老人觉得冷（24℃），年轻人觉得热（24℃）；
• 故障'藏猫猫'：设备坏了全靠业主投诉，维修平均要 2 小时（如空调传感器故障导致温度显示不准，3 天才发现）。

项目目标很明确：用 Java 大数据改造，实现'实时响应、个性调节、主动预判'，解决上述所有痛点。

4.2 项目核心效果（数据对比，均来自物业报告）

数据可信度说明：上述数据由物业每月抽样 100 户统计，结合设备后台日志计算得出，2024 年 Q3 报告已在小区公告栏公示（可联系物业查阅）。其中'舒适度达标率'按 GB/T 50785-2012《民用建筑室内热湿环境评价标准》计算（温度 18-28℃+ 湿度 40-60%+PM2.5≤50μg/m³ 即为达标）。

4.3 典型场景：'老人房的无感智能'

2024 年 9 月 10 日，7 栋 201 室张阿姨（68 岁，独居）的使用场景，是项目'个性化调节'的最佳诠释——技术不应该是复杂的操作，而应该是'润物细无声'的关怀。

4.3.1 场景触发：多维度数据融合

当天系统采集到的信息（来自设备日志）：

• 实时环境：老人房传感器每 2 秒上报'温度 22℃，湿度 38%'（连续 5 分钟稳定）；
• 用户偏好：张阿姨在 APP 设置'日常温度 25℃，湿度 45%，22:00 自动开启睡眠模式（23℃）'；
• 行为习惯：历史数据显示她每天 14:00-15:00 在卧室看电视（窗帘关闭，光照强度 < 300lux）；
• 外部数据：气象局 API 返回'当天下午降温，室外温度从 25℃降至 20℃，风力 3 级'。

4.3.2 决策执行：规则链联动

Spark Streaming 计算出'老人房平均温度 22℃（低于偏好 3℃），湿度 38%（低于偏好 7%）'，异常级别'medium'，决策引擎按优先级触发 2 条规则：

• 规则 2（温度调节）：空调切换至制热模式，目标 25℃（考虑室外降温，提前预热）；
• 规则 3（湿度调节）：加湿器开启，目标 45%（避免升温后空气干燥）。

4.3.3 用户反馈与效果

张阿姨 14:00 走进卧室时，温度已升至 24.5℃，湿度 43%——她在 9 月业主满意度调查中写道：'现在不用找遥控器，房间永远暖暖的，湿度也刚好，比儿女还贴心。'

这个场景在改造前，张阿姨需要弯腰找空调遥控器（她有腰疾），调完温度还要开加湿器，经常记不清操作步骤；改造后完全'无感'，这就是技术落地的'温度'。

五、Java 大数据在智能家居中的应用拓展

做完三个项目，越来越觉得：智能家居的终极形态是'让人忘记技术的存在'。结合行业趋势和业主需求，Java 大数据还有三个值得深耕的方向，也是下一个项目的重点。

5.1 融合 AI 的个性化学习（从'规则驱动'到'数据驱动'）

目前的决策引擎依赖'人工配置规则'，未来可以结合 Spark MLlib 让系统'自己学用户习惯'。比如张阿姨冬天喜欢把温度调高 1℃，系统应该记住这个偏好，无需人工改规则。

5.1.1 数据采集与预处理（MySQL 表设计 + Spark SQL）

• 预处理 SQL（提取用户调节偏好特征）：

用户调节记录表示例（user_adjust_record）：

-- 近 3 个月用户在不同场景下的平均调节幅度
SELECT user_id, room_id, weather, is_sleep,
       AVG(temp_after - temp_before) AS temp_adjust_delta, -- 温度调节幅度（正数=调高，负数=调低）
       AVG(hum_after - hum_before) AS hum_adjust_delta -- 湿度调节幅度
FROM user_adjust_record 
WHERE adjust_time >= DATE_SUB(CURDATE(), 90) -- 取近 3 个月数据，避免过旧习惯影响
GROUP BY user_id, room_id, weather, is_sleep;

5.1.2 模型训练（Spark MLlib ALS 算法，附完整代码）

package com.smarthome.ai;

import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.recommendation.ALS;
import org.apache.spark.ml.recommendation.ALSModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 用户偏好推荐模型（基于 ALS 协同过滤算法，C 市项目二期规划）
 * 功能：预测用户在不同场景（天气/时段/房间）下的温度/湿度偏好调节幅度
 */
public class UserPreferenceModel {
    private static final Logger log = LoggerFactory.getLogger(UserPreferenceModel.class);
    
    public static void main(String[] args) {
        // 1. 初始化 SparkSession（YARN 集群模式）
        SparkSession spark = SparkSession.builder()
            .appName("UserPreferenceModel-Training")
            .master("yarn")
            .config("spark.executor.memory", "8g") // 训练需要更多内存
            .config("spark.driver.memory", "4g")
            .getOrCreate();
        
        try {
            // 2. 加载预处理后的特征数据（来自 HDFS）
            Dataset<Row> features = spark.read()
                .option("header", true)
                .option("inferSchema", true)
                .csv("hdfs:///smarthome/ai/features/user_adjust_features.csv");
            
            // 3. 特征编码（将字符串 ID 转为整数索引，ALS 算法要求）
            Dataset<Row> encodedFeatures = features
                // 用户 ID 哈希为整数（避免原 ID 过长）
                .withColumn("user_idx", functions.hash(functions.col("user_id")).mod(10000))
                // 场景（房间 + 天气 + 时段）哈希为整数
                .withColumn("scene_idx", functions.hash(
                    functions.concat("room_id", "weather", "is_sleep")
                ).mod(1000))
                // 选择训练所需字段（用户索引、场景索引、温度调节幅度作为标签）
                .select(
                    functions.col("user_idx").cast("integer"),
                    functions.col("scene_idx").cast("integer"),
                    functions.col("temp_adjust_delta").cast("double").as("label")
                )
                .na().drop(); // 过滤空值
            
            // 4. 划分训练集和测试集（8:2）
            Dataset<Row>[] splits = encodedFeatures.randomSplit(new double[]{0.8, 0.2}, 42); // 固定随机种子，结果可复现
            Dataset<Row> training = splits[0];
            Dataset<Row> test = splits[1];
            
            // 5. 训练 ALS 模型（协同过滤推荐算法，适合用户 - 场景偏好预测）
            ALS als = new ALS()
                .setMaxIter(10) // 迭代次数（10 次平衡效果与性能）
                .setRegParam(0.01) // 正则化参数（防止过拟合）
                .setUserCol("user_idx") // 用户索引列
                .setItemCol("scene_idx") // 场景索引列
                .setRatingCol("label") // 标签列（温度调节幅度）
                .setColdStartStrategy("drop"); // 冷启动策略（新用户/场景直接丢弃）
            
            ALSModel model = als.fit(training);
            
            // 6. 模型评估（RMSE 越小越好，目标<0.5℃）
            Dataset<Row> predictions = model.transform(test);
            RegressionEvaluator evaluator = new RegressionEvaluator()
                .setMetricName("rmse") // 均方根误差
                .setLabelCol("label")
                .setPredictionCol("prediction");
            double rmse = evaluator.evaluate(predictions);
            log.info("模型评估结果：RMSE = {}℃（越小越精准，目标<0.5℃）", rmse);
            
            // 7. 保存模型（供决策引擎实时调用）
            String modelPath = "hdfs:///smarthome/ai/models/user_preference_model_v1/";
            model.save(modelPath);
            log.info("模型训练完成，已保存至：{}", modelPath);
        } catch (Exception e) {
            log.error("模型训练失败", e);
        } finally {
            spark.stop();
        }
    }
}

5.1.3 模型与决策引擎集成（关键代码补充）

package com.smarthome.service.impl;

import org.apache.spark.ml.recommendation.ALSModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import com.smarthome.entity.RoomMetric;
import com.smarthome.entity.UserPreference;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

/**
 * AI 偏好预测服务（集成 ALS 模型到决策引擎）
 */
@Component
public class AiPreferenceService {
    private final ALSModel alsModel;
    private final SparkSession spark;
    
    // 初始化：加载预训练模型
    public AiPreferenceService(@Value("${ai.model.path}") String modelPath) {
        this.spark = SparkSession.builder()
            .appName("AiPreferenceService")
            .master("local[2]") // 本地模式，用 2 核（决策引擎部署在同一服务器）
            .getOrCreate();
        this.alsModel = ALSModel.load(modelPath);
        log.info("✅ AI 偏好预测模型加载完成，路径：{}", modelPath);
    }
    
    /**
     * 预测用户在当前场景下的温度调节幅度
     */
    public double predictTempAdjustDelta(RoomMetric metric, UserPreference pref, String weather) {
        try {
            // 1. 构造当前场景特征
            String userId = pref.getUserId();
            String roomId = metric.getRoomId();
            int isSleep = isSleepTime() ? 1 : 0; // 判断是否睡眠时段
            
            // 2. 特征编码（与训练时一致）
            int userIdx = Math.abs(userId.hashCode() % 10000);
            String sceneKey = roomId + weather + isSleep;
            int sceneIdx = Math.abs(sceneKey.hashCode() % 1000);
            
            // 3. 构造 Spark DataFrame（模型输入格式）
            Map<String, Object> data = new HashMap<>();
            data.put("user_idx", userIdx);
            data.put("scene_idx", sceneIdx);
            Dataset<Row> sceneData = spark.createDataFrame(Arrays.asList(data),
                new StructType().add("user_idx", DataTypes.IntegerType).add("scene_idx", DataTypes.IntegerType));
            
            // 4. 调用模型预测调节幅度
            Dataset<Row> prediction = alsModel.transform(sceneData);
            if (prediction.count() == 0) {
                return 0.0; // 无预测结果，用默认值
            }
            double delta = prediction.select("prediction").first().getDouble(0);
            
            // 5. 限制调节幅度（避免预测异常值，最大±2℃）
            return Math.max(-2.0, Math.min(2.0, delta));
        } catch (Exception e) {
            log.error("温度调节幅度预测失败", e);
            return 0.0; // 失败时用默认值，不影响主流程
        }
    }
    
    // 判断是否睡眠时段（22:00-6:00）
    private boolean isSleepTime() {
        int hour = java.time.LocalTime.now().getHour();
        return hour >= 22 || hour <= 6;
    }
}

落地效果预期：模型部署后，决策引擎会结合'规则结果 + AI 预测'调节设备。比如张阿姨在'雨天 + 卧室 + 非睡眠'场景，模型预测'温度调节幅度 + 1.5℃'，系统会在规则目标 25℃的基础上加 1.5℃，更贴合她的习惯——这一步能减少 30% 的用户手动调节（基于前期小范围测试）。

5.2 跨空间的智能联动（从'单户'到'小区 - 家庭'）

未来可打通'小区公共区域'与'家庭内部'数据，实现更自然的联动。比如：

5.2.1 场景 1：小区车库联动（技术方案）

• 触发条件：业主车辆进入车库（通过车牌识别系统）；
• 联动逻辑：
  • 物业系统通过Spring Cloud Stream将'车牌 + 进入时间'推送到 Kafka 主题community/garage/entry；
  • 决策引擎消费消息，查询'车牌 - 业主 - 房间'映射表，获取用户 ID 和预计到家时间（车库到家门约 2 分钟）；
  • 提前 1 分钟启动客厅空调（按用户偏好温度），打开玄关灯。
• 核心规则（新增到smarthome.drl）：

rule "Garage Entry Trigger Home Preheat"
    salience 11
    when
        $car: CarEntryEvent(status == "ENTER") // 车库进入事件
        $pref: UserPreference(userId == $car.getUserId()) // 计算预计到家时间（当前时间 +2 分钟）
        $arriveTime: Long() from (System.currentTimeMillis() + 2*60*1000)
    then
        // 提前 1 分钟启动设备
        schedulerService.schedule(() -> {
            deviceControlService.controlAircon("livingroom", "AUTO", $pref.getTempPreference());
            deviceControlService.controlLight("entrance", "ON");
        }, $arriveTime - 60*1000);
        log.info("规则触发 [车库联动]：用户{}车辆进入，将提前预热客厅", $car.getUserId());
end

5.2.2 场景 2：公共绿化联动（解决实际问题）

小区绿化灌溉时，室外湿度常达 80% 以上，容易导致室内返潮——通过联动可自动关窗除湿：

• 数据来源：小区绿化系统的'灌溉计划'（定时任务）+ 室外气象站的湿度数据；
• 联动逻辑：灌溉前 10 分钟，自动关闭一楼住户的窗户，开启除湿模式。

5.3 低碳节能与能源优化（响应'双碳'政策）

结合国家'双碳'目标，Java 大数据可帮助家庭降低能耗：

5.3.1 峰谷电价优化（经济 + 环保）

• 数据基础：电网峰谷电价表（如 00:00-08:00 为谷段，电价 0.3 元 / 度；18:00-22:00 为峰段，0.8 元 / 度）；
• 优化逻辑：Spark 分析家庭用电习惯（如热水器日均用 3 度电），在谷段自动加热至 70℃，峰段不加热，每天可省 1.5 元（3 度×(0.8-0.3)）。

5.3.2 能耗异常告警（帮用户省钱）

• 检测逻辑：实时监测设备功率，若空调连续 24 小时运行（非制热 / 制冷季）、热水器温度设 80℃以上，推送给用户：'您家空调已连续运行 24 小时，关闭可省 3.2 度电（约 2.5 元）'；
• 数据支撑：基于项目数据，此类优化可降低家庭总能耗 15%-20%。

结束语：技术的温度，藏在细节里

做第一个项目时，总想着'用最炫的技术'，把架构图画得花里胡哨；到第三个项目，更关注'张阿姨会不会觉得冷''业主调温度是否方便'——技术的价值，最终要落到人的体验上。

有人问：'Python 做 AI 更方便，为什么坚持用 Java？'答案很简单：智能家居是'7×24 小时不能停'的系统，Java 的稳定性、生态成熟度，是凌晨 3 点空调不宕机的保障。当然，也会用 Python 做 AI 模型训练，但生产环境的决策和控制，Java 永远是'定海神针'。

技术的迭代，往往从用户的一个'不爽'开始。