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

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

• 运行数据采集模块：

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，Body 如下：

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

Headers 加 Authorization：Basic YWRtaW46cGFzc3dvcmQ=。

![系统架构图]

正文

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

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

智能家居设备来自不同厂商，数据格式、传输协议差异极大。

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

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

以一个 3 室 2 厅的家庭为例，若部署 15 个智能设备，数据规模如下：

• 实时数据：日均 100MB。
• 历史数据：按保存 1 年计算，约 36GB。
• 结构化数据：用户配置、设备属性等约 50MB。

传统的单机数据库（如 SQLite）根本扛不住，单表数据超 1000 万条时查询延迟严重。

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

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

理想的状态是系统提前 10 分钟启动空调，用户进门就能享受到'刚刚好'的温度。初期用'Java 定时器 + MySQL 查询'做决策，响应延迟 8 秒；后来换成 Spark Streaming，把延迟降到 2 秒内。

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

传统智能家居的调节策略靠'固定阈值'，但忽略了'环境联动'和'用户习惯'。例如同样是 26℃，湿度 60% 和湿度 30% 的体感完全不同。

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

环境监测涉及大量用户隐私数据，很多平台存在'数据明文传输''存储不安全'的问题。所有数据传输需加 SSL 加密，存储用脱敏。

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

2.1 技术栈选型对比与决策

2.2 系统整体架构

架构采用纵向布局，分为'感知 - 传输 - 处理 - 存储 - 应用 - 执行'分层。传输层用 EMQX 作为 MQTT Broker，存储层用 Cassandra 存时序数据。

![系统架构图]

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

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

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

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.smarthome</groupId>
    <artifactId>data-collect-module</artifactId>
    <version>1.0.0</version>
    <dependencies>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.5</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.41</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.0</version>
        </dependency>
    </dependencies>
</project>

3.1.2 MQTT 数据采集客户端

package com.smarthome.collect;
import org.eclipse.paho.client.mqttv3.*;
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/#"};
    private MqttClient mqttClient;

    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);
        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);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String rawData = new String(message.getPayload(), "UTF-8");
        // 转发到 Kafka 逻辑
    }

    @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();
        }
    }

    public static void main(String[] args) {
        MqttDataCollector collector = new MqttDataCollector();
        try {
            collector.init();
            Thread.currentThread().join();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.1.3 Kafka 生产者工具类

package com.smarthome.collect;
import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaProducer {
    private static final Logger log = LoggerFactory.getLogger(KafkaProducer.class);
    private KafkaProducer<String, String> producer;

    public KafkaProducer(String brokers) {
        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());
        this.producer = new KafkaProducer<>(props);
    }

    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 消息发送失败", exception);
            }
        });
    }

    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;

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("sensor_hum", "humidity");
    }

    public static JSONObject standardize(String deviceType, String rawData) {
        try {
            JSONObject rawJson = JSONObject.parseObject(rawData);
            JSONObject standardJson = new JSONObject();
            standardJson.put("device_id", getStandardField(rawJson, deviceType, "device_id"));
            standardJson.put("collect_time", System.currentTimeMillis());
            standardJson.put("device_type", deviceType);
            return standardJson;
        } catch (Exception e) {
            log.error("数据标准化失败", e);
            return null;
        }
    }

    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();
            }
        }
        return "UNKNOWN_DEVICE_" + System.currentTimeMillis();
    }
}

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

3.2.1 核心依赖配置（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>
</dependencies>

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.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.KafkaUtils;
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;
    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");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(BATCH_DURATION));
        jssc.checkpoint("hdfs:///smarthome/checkpoint/streaming");

        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");

        Collection<String> topics = Collections.singletonList(KAFKA_TOPIC_STANDARD);
        JavaDStream<String> kafkaDStream = KafkaUtils.createDirectStream(
            jssc,
            KafkaUtils.LocationStrategies.PreferConsistent(),
            KafkaUtils.ConsumerStrategies.Subscribe(topics, kafkaParams)
        ).map(record -> {
            try {
                return JSONObject.parseObject(record.value());
            } catch (Exception e) {
                return null;
            }
        }).filter(Objects::nonNull);

        kafkaDStream.print();
        jssc.start();
        jssc.awaitTermination();
    }
}

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

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

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.drools</groupId>
        <artifactId>drools-core</artifactId>
        <version>8.44.0.Final</version>
    </dependency>
    <dependency>
        <groupId>org.kie</groupId>
        <artifactId>kie-spring</artifactId>
        <version>8.44.0.Final</version>
    </dependency>
</dependencies>

3.3.2 Drools 决策规则配置

规则文件放在 src/main/resources/rules/smarthome.drl。

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

global DeviceControlService deviceControlService;

rule "PM2.5 Severe Anomaly"
salience 15
when
    $metric: RoomMetric(anomaly_level == "high", pm25_anomaly == true, maxPm25 > 100)
then
    deviceControlService.controlPurifier($metric.getRoomId(), "ON", 5);
end

rule "Temperature Anomaly"
salience 10
when
    $metric: RoomMetric(anomaly_level in ("medium", "high"), temp_anomaly == true)
then
    deviceControlService.controlAircon($metric.getRoomId(), "COOL", 24);
end

3.3.3 Drools 与 Spring Boot 集成配置

package com.smarthome.config;
import org.kie.api.KieServices;
import org.kie.api.builder.KieBuilder;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class DroolsConfig {
    @Bean
    public KieContainer kieContainer() throws IOException {
        KieServices kieServices = KieServices.Factory.get();
        KieFileSystem kieFileSystem = kieServices.newKieFileSystem();
        // 加载规则文件逻辑
        KieBuilder kieBuilder = kieServices.newKieBuilder(kieFileSystem);
        kieBuilder.buildAll();
        return kieServices.newKieContainer(kieRepository.getDefaultReleaseId());
    }

    @Bean
    public KieSession kieSession() throws IOException {
        return kieContainer().newKieSession();
    }
}

3.3.4 决策引擎核心服务

package com.smarthome.service.impl;
import com.smarthome.entity.RoomMetric;
import com.smarthome.service.DecisionService;
import org.kie.api.runtime.KieSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

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

    @Override
    public void executeDecision(RoomMetric metric) {
        kieSession.insert(metric);
        kieSession.fireAllRules();
        kieSession.clear();
    }
}

3.3.5 设备控制服务实现

package com.smarthome.service.impl;
import com.smarthome.mqtt.MqttPublisher;
import com.smarthome.service.DeviceControlService;
import com.alibaba.fastjson2.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

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

    @Override
    public void controlAircon(String roomId, String mode, double targetTemp) {
        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));
        String topic = "smarthome/control/" + roomId + "/aircon";
        mqttPublisher.publish(topic, cmd.toJSONString());
    }
}

3.3.6 MQTT 发布器工具类

package com.smarthome.mqtt;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class MqttPublisher {
    @Value("${mqtt.broker}")
    private String broker;
    private MqttClient mqttClient;

    public void publish(String topic, String payload) {
        try {
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(1);
            mqttClient.publish(topic, message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }
}

3.3.7 核心实体类

package com.smarthome.entity;
import lombok.Data;

@Data
public class RoomMetric {
    private String roomId;
    private double avgTemperature;
    private double avgHumidity;
    private double maxPm25;
    private String anomalyLevel;
}

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

4.1 项目背景与改造前痛点

该小区预装了传统智能家居系统，业主投诉集中在设备'各玩各的'、响应'慢吞吞'、不懂'看人下菜'、故障'藏猫猫'。

4.2 项目核心效果

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

系统采集实时环境、用户偏好、行为习惯及外部数据。Spark Streaming 计算出异常级别，决策引擎按优先级触发规则，自动调节空调和加湿器，无需用户干预。

![典型场景示意图]

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

5.1 融合 AI 的个性化学习

结合 Spark MLlib 让系统'自己学用户习惯'。使用 ALS 协同过滤算法预测用户在不同场景下的温度/湿度偏好调节幅度。

// Spark MLlib ALS 模型训练示例
val als = new ALS()
  .setMaxIter(10)
  .setRegParam(0.01)
  .setUserCol("user_idx")
  .setItemCol("scene_idx")
  .setRatingCol("label")
val model = als.fit(training)

5.2 跨空间的智能联动

打通'小区公共区域'与'家庭内部'数据。例如车库进入联动客厅预热，绿化灌溉联动室内除湿。

5.3 低碳节能与能源优化

结合峰谷电价优化用电习惯，检测能耗异常告警，降低家庭总能耗 15%-20%。

![低碳节能示意图]