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

<?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 客户端 -->
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.5</version>
        </dependency>
        <!-- 2. JSON 解析 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.41</version>
        </dependency>
        <!-- 3. Kafka 客户端 -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.0</version>
        </dependency>
    </dependencies>
</project>

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;

public class MqttDataCollector implements MqttCallback {
    private static final Logger log = LoggerFactory.getLogger(MqttDataCollector.class);
    
    private static final String MQTT_BROKER = "ssl://emqx.smarthome.com:8883";
    private static final String MQTT_CLIENT_ID = "java-collector-" + System.currentTimeMillis();
    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/sensor/#",
        "smarthome/device/purifier/#"
    };
    
    private MqttClient mqttClient;
    private KafkaProducer kafkaProducer;

    public void init() throws MqttException {
        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(MQTT_BROKER, MQTT_CLIENT_ID, persistence);
        
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setUserName(MQTT_USERNAME);
        connOpts.setPassword(MQTT_PASSWORD.toCharArray());
        connOpts.setAutomaticReconnect(true);
        connOpts.setConnectionTimeout(10);
        connOpts.setKeepAliveInterval(30);
        connOpts.setCleanSession(false);
        
        Properties sslProps = new Properties();
        sslProps.put("ssl.trustStore", "src/main/resources/cert/emqx-truststore.jks");
        sslProps.put("ssl.trustStorePassword", "truststore123");
        connOpts.setSSLProperties(sslProps);
        
        mqttClient.setCallback(this);
        mqttClient.connect(connOpts);
        for (String topic : SUBSCRIBE_TOPICS) {
            mqttClient.subscribe(topic, 1);
            log.info("✅ 订阅设备主题成功：{}", topic);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String rawData = new String(message.getPayload(), "UTF-8");
        try {
            kafkaProducer.send("smarthome_raw_data", topic, rawData);
        } catch (Exception e) {
            log.error("❌ 设备消息转发 Kafka 失败", e);
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("⚠️ MQTT 连接断开，正在自动重连...", cause);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {}
}

数据标准化服务

通过字段映射表统一不同厂商的数据格式。

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;

public class DataStandardizer {
    private static final Logger log = LoggerFactory.getLogger(DataStandardizer.class);
    private static final Map<String, String> FIELD_MAPPING = new HashMap<>();

    static {
        // 示例：空调字段映射
        FIELD_MAPPING.put("aircon_temp", "temperature");
        FIELD_MAPPING.put("aircon_wind_speed", "fan_level");
        // 净化器字段映射
        FIELD_MAPPING.put("purifier_pm25", "pm2_5");
        // 传感器字段映射
        FIELD_MAPPING.put("sensor_temp", "temperature");
        FIELD_MAPPING.put("sensor_hum", "humidity");
    }

    public static JSONObject standardize(String deviceType, String rawData) {
        try {
            JSONObject rawJson = JSONObject.parseObject(rawData);
            if (rawJson == null) return null;
            
            JSONObject standardJson = new JSONObject();
            standardJson.put("device_id", getStandardField(rawJson, deviceType));
            standardJson.put("collect_time", System.currentTimeMillis());
            standardJson.put("device_type", deviceType);
            standardJson.put("raw_data", rawData);

            switch (deviceType) {
                case "aircon":
                    standardizeAircon(rawJson, standardJson);
                    break;
                case "purifier":
                    standardizePurifier(rawJson, standardJson);
                    break;
                case "sensor":
                    standardizeSensor(rawJson, standardJson);
                    break;
            }
            return standardJson;
        } catch (Exception e) {
            log.error("❌ 数据标准化失败", e);
            return null;
        }
    }

    private static void standardizeAircon(JSONObject rawJson, JSONObject standardJson) {
        standardJson.put("temperature", getDoubleValue(rawJson, "aircon", "temp", "temp_val"));
        standardJson.put("humidity", getDoubleValue(rawJson, "aircon", "hum", "humidity"));
        standardJson.put("work_mode", getStringValue(rawJson, "aircon", "mode"));
    }

    private static double getDoubleValue(JSONObject rawJson, String deviceType, String... fields) {
        for (String field : fields) {
            String key = FIELD_MAPPING.getOrDefault(deviceType + "_" + field, field);
            Double val = rawJson.getDouble(key);
            if (val != null && !Double.isNaN(val)) {
                return Math.round(val * 10) / 10.0;
            }
        }
        return 0.0;
    }

    private static String getStringValue(JSONObject rawJson, String deviceType, String... fields) {
        for (String field : fields) {
            String key = FIELD_MAPPING.getOrDefault(deviceType + "_" + field, field);
            String val = rawJson.getString(key);
            if (val != null && !val.trim().isEmpty()) {
                return val.trim();
            }
        }
        return "UNKNOWN";
    }

    private static String getStandardField(JSONObject rawJson, String deviceType) {
        String val = rawJson.getString("device_id");
        if (val != null && !val.trim().isEmpty()) return val.trim();
        return "UNKNOWN_DEVICE_" + System.currentTimeMillis();
    }
}

2. Spark Streaming 实时数据处理

采集后的数据需实时计算环境指标，为决策引擎提供依据。

核心依赖配置 (pom.xml)

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.12</artifactId>
        <version>3.5.0</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
        <version>3.5.0</version>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson2</artifactId>
        <version>2.0.41</version>
    </dependency>
</dependencies>

实时处理代码

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.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSONObject;

import java.util.*;

public class SmartHomeStreamProcessor {
    private static final Logger log = LoggerFactory.getLogger(SmartHomeStreamProcessor.class);
    private static final int BATCH_DURATION = 5; // 微批间隔 5 秒
    private static final String KAFKA_TOPIC_STANDARD = "smarthome_standard_data";

    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf()
            .setAppName("SmartHome-Stream-Processor")
            .setMaster("yarn")
            .set("spark.executor.instances", "3")
            .set("spark.executor.memory", "4g")
            .set("spark.streaming.backpressure.enabled", "true");

        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(BATCH_DURATION));
        jssc.checkpoint("hdfs:///smarthome/checkpoint/streaming");

        try {
            Map<String, Object> kafkaParams = new HashMap<>();
            kafkaParams.put("bootstrap.servers", "kafka.smarthome.com:9092");
            kafkaParams.put("group.id", "spark-streaming-group");
            kafkaParams.put("auto.offset.reset", "latest");
            kafkaParams.put("enable.auto.commit", false);

            Collection<String> topics = Collections.singletonList(KAFKA_TOPIC_STANDARD);
            JavaDStream<String> kafkaDStream = KafkaUtils.createDirectStream(
                jssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topics, kafkaParams)
            ).map(record -> record.value());

            JavaDStream<JSONObject> validDataStream = kafkaDStream
                .map(rawJson -> {
                    try { return JSONObject.parseObject(rawJson); }
                    catch (Exception e) { return null; }
                })
                .filter(Objects::nonNull)
                .filter(json -> json.getString("device_id") != null)
                .map(json -> {
                    String deviceId = json.getString("device_id");
                    String roomId = deviceId.split("_")[0];
                    json.put("room_id", roomId);
                    return json;
                });

            JavaDStream<JSONObject> roomMetricStream = validDataStream
                .mapToPair(json -> new scala.Tuple2<>(json.getString("room_id"), json))
                .groupByKey()
                .mapValues(jsons -> {
                    JSONObject metric = new JSONObject();
                    double tempSum = 0.0, humSum = 0.0;
                    int count = 0;
                    for (JSONObject json : jsons) {
                        if (json.containsKey("temperature")) {
                            tempSum += json.getDoubleValue("temperature");
                            count++;
                        }
                        if (json.containsKey("humidity")) {
                            humSum += json.getDoubleValue("humidity");
                        }
                    }
                    metric.put("avg_temperature", count > 0 ? Math.round(tempSum / count * 10) / 10.0 : 0.0);
                    metric.put("avg_humidity", Math.round(humSum / count * 10) / 10.0);
                    metric.put("calculate_time", System.currentTimeMillis());
                    detectEnvironmentAnomaly(metric);
                    return metric;
                });

            roomMetricStream.print(5);
            jssc.start();
            jssc.awaitTermination();
        } finally {
            jssc.stop(true, true);
        }
    }

    private static void detectEnvironmentAnomaly(JSONObject metric) {
        double avgTemp = metric.getDoubleValue("avg_temperature");
        double avgHum = metric.getDoubleValue("avg_humidity");
        boolean tempAnomaly = avgTemp < 16.0 || avgTemp > 30.0;
        boolean humAnomaly = avgHum < 30.0 || avgHum > 70.0;
        
        String anomalyLevel = "low";
        if (tempAnomaly && humAnomaly) anomalyLevel = "high";
        else if (tempAnomaly || humAnomaly) anomalyLevel = "medium";
        
        metric.put("anomaly_level", anomalyLevel);
        if (!"low".equals(anomalyLevel)) {
            log.warn("⚠️ 房间 {} 环境异常 | 级别：{}", metric.getString("room_id"), anomalyLevel);
        }
    }
}

3. 智能决策引擎与设备控制（Spring Boot+Drools）

决策引擎负责根据环境指标触发规则并下发控制指令。

Drools 决策规则配置

规则文件 smarthome.drl 定义业务逻辑。

package com.smarthome.rules;
import com.smarthome.entity.RoomMetric;
import com.smarthome.service.DeviceControlService;

global DeviceControlService deviceControlService;

rule "PM2.5 Severe Anomaly - Turn On Purifier High"
    salience 15
when
    $metric: RoomMetric(anomaly_level == "high", pm25_anomaly == true)
then
    deviceControlService.controlPurifier($metric.getRoomId(), "ON", 5);
    log.info("规则触发 [PM2.5 紧急处理]：房间{}PM2.5 严重超标", $metric.getRoomId());
end

rule "Temperature Anomaly - Adjust Aircon"
    salience 10
when
    $metric: RoomMetric(anomaly_level in ("medium", "high"), temp_anomaly == true)
then
    String mode = $metric.getAvgTemperature() > 24 ? "COOL" : "HEAT";
    deviceControlService.controlAircon($metric.getRoomId(), mode, 24.0);
    log.info("规则触发 [温度调节]：房间{}当前{}℃", $metric.getRoomId(), $metric.getAvgTemperature());
end

rule "Multi-Device 联动 - Temp High + Hum High"
    salience 12
when
    $metric: RoomMetric(avgTemperature > 26, avgHumidity > 65)
then
    deviceControlService.controlAircon($metric.getRoomId(), "DEHUMIDIFY", 24);
    log.info("规则触发 [温湿双高联动]：房间{}开启除湿模式", $metric.getRoomId());
end

决策引擎核心服务

package com.smarthome.service.impl;

import com.smarthome.entity.RoomMetric;
import com.smarthome.entity.UserPreference;
import org.kie.api.runtime.KieSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class DecisionServiceImpl {
    @Autowired
    private KieSession kieSession;

    public void executeDecision(RoomMetric metric) {
        try {
            UserPreference userPreference = getUserPreference(metric.getRoomId());
            kieSession.insert(metric);
            kieSession.insert(userPreference);
            kieSession.fireAllRules();
            log.info("📊 房间{}决策完成", metric.getRoomId());
        } catch (Exception e) {
            log.error("❌ 决策执行失败", e);
        } finally {
            kieSession.clear();
        }
    }

    private UserPreference getUserPreference(String roomId) {
        // 默认偏好
        return UserPreference.builder()
            .userId(roomId.replace("room", "user"))
            .tempPreference(24.0)
            .build();
    }
}

设备控制服务

package com.smarthome.service.impl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.smarthome.mqtt.MqttPublisher;

@Service
public class DeviceControlServiceImpl {
    @Autowired
    private MqttPublisher mqttPublisher;

    public void controlAircon(String roomId, String mode, double targetTemp) {
        String topic = "smarthome/control/" + roomId + "/aircon";
        mqttPublisher.publish(topic, "{\"cmd\": \"AIRCON_CONTROL\", \"mode\": \"" + mode + "\"}");
    }

    public void controlPurifier(String roomId, String status, int fanLevel) {
        String topic = "smarthome/control/" + roomId + "/purifier";
        mqttPublisher.publish(topic, "{\"cmd\": \"PURIFIER_CONTROL\", \"status\": \"" + status + "\"}");
    }
}

实战案例复盘

在某高端小区项目中，系统上线后稳定运行 6 个月，主要效果如下：

典型场景：老人房无感智能。系统结合用户偏好（日常 25℃）、历史习惯（看电视时关窗帘）及外部天气数据，自动调节温湿度。老人无需操作遥控器，房间始终保持在舒适状态。

未来拓展方向

1. 融合 AI 的个性化学习

引入 Spark MLlib 进行协同过滤推荐，预测用户在不同场景下的偏好调节幅度，减少人工规则配置。

2. 跨空间智能联动

打通小区公共区域与家庭内部数据。例如，车辆进入车库时，提前预热客厅空调；绿化灌溉时，自动关闭一楼窗户防返潮。

3. 低碳节能优化

结合峰谷电价政策，在谷段自动加热热水器，峰段停止加热，降低家庭总能耗 15%-20%。

结语

智能家居的终极形态是让人忘记技术的存在。Java 大数据技术栈凭借其稳定性与生态成熟度，能够有效解决设备异构、数据孤岛及决策滞后等痛点。通过 MQTT、Spark Streaming 与 Drools 的组合，不仅能实现毫秒级响应，还能通过规则与算法的结合，提供真正懂用户的智慧体验。