Java 大数据赋能智能家居：设备联动与节能实践

1.3.2 联动规则实体类（对应 MySQL 配置表）

package com.smarthome.entity;
import lombok.Data;
import java.io.Serializable;

@Data
public class LinkageRule implements Serializable {
    private Long ruleId; // 规则 ID
    private String ruleName; // 规则名称
    private String conditionSql; // 触发条件
    private String actionJson; // 执行动作
    private int isEnable; // 是否启用
    private String sceneType; // 场景类型
    private String userId; // 所属用户 ID
    private String createTime; // 创建时间
    private String updateTime; // 更新时间
}

1.3.3 缺失工具类补充：SpringContextUtil

package com.smarthome.util;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

@Component
public class SpringContextUtil implements ApplicationContextAware {
    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext context) throws BeansException {
        applicationContext = context;
    }

    public static <T> T getBean(Class<T> clazz) {
        if (applicationContext == null) throw new RuntimeException("SpringContext 未初始化");
        try { return applicationContext.getBean(clazz); }
        catch (Exception e) { throw new RuntimeException("获取 Bean 失败", e); }
    }
}

二、核心场景 1：动态联动引擎 —— 从'固定规则'到'数据驱动'

2.1 行业痛点

传统联动系统存在规则刚性、无上下文感知、跨品牌兼容差三大问题。调研显示，跨品牌设备联动率不足 35%，平均延迟 3.2 秒。

2.2 解决方案：Flink SQL 驱动的动态联动引擎

基于 Flink 构建'状态流 + 广播规则流'的联动引擎，核心逻辑是'设备状态实时感知 + 联动规则动态更新 + 多条件智能匹配'。

2.2.1 核心依赖（pom.xml 关键配置）

<dependencies>
    <!-- Flink 核心依赖 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!-- Calcite SQL 引擎 -->
    <dependency>
        <groupId>org.apache.calcite</groupId>
        <artifactId>calcite-core</artifactId>
        <version>${calcite.version}</version>
    </dependency>
    <!-- MQTT 客户端 -->
    <dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>${mqtt.version}</version>
    </dependency>
</dependencies>

2.2.2 关键工具类：KafkaSourceBuilder

package com.smarthome.source;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;

public class KafkaSourceBuilder {
    public static <T> DataStream<T> build(StreamExecutionEnvironment env, String topic, String groupId, DeserializationSchema<T> deserializer) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-node1:9092,kafka-node2:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topic, deserializer, props);
        return env.addSource(kafkaConsumer).name("Kafka-Source-" + topic).uid("kafka-source-" + topic);
    }
}

2.2.3 关键工具类：DeviceControlSink

package com.smarthome.sink;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class DeviceControlSink extends RichSinkFunction<String> {
    private final String brokerUrl;
    private MqttClient mqttClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setAutomaticReconnect(true);
        mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
        mqttClient.connect(connOpts);
    }

    @Override
    public void invoke(String controlCmd, Context context) throws Exception {
        if (!mqttClient.isConnected()) throw new RuntimeException("MQTT 连接已断开");
        String topic = "device/control/" + extractDeviceId(controlCmd);
        MqttMessage message = new MqttMessage(controlCmd.getBytes("UTF-8"));
        message.setQos(1);
        mqttClient.publish(topic, message);
    }
}

2.2.4 动态联动核心 Job

package com.smarthome.flink.job;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.BroadcastStream;

public class DeviceLinkageJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(180000);
        env.setParallelism(12);

        DataStream<DeviceStatus> deviceStatusStream = KafkaSourceBuilder.build(env, "device_status_topic", "group1", new SimpleStringSchema());
        DataStream<LinkageRule> ruleStream = KafkaSourceBuilder.build(env, "linkage_rule_cdc_topic", "group2", new SimpleStringSchema());

        BroadcastStream<LinkageRule> broadcastRuleStream = ruleStream.broadcast(new MapStateDescriptor<>("rule-state", String.class, LinkageRule.class));

        DataStream<String> controlStream = deviceStatusStream.connect(broadcastRuleStream).process(new BroadcastProcessFunction<>());
        controlStream.addSink(new DeviceControlSink("ssl://mqtt-broker:8883"));

        env.execute("Device Linkage Job");
    }
}

2.3 真实案例：北京望京 SOHO 公寓'起床场景'动态联动

2.3.1 需求背景

用户希望每天早上 7 点起床时，窗帘拉开、空调切换舒适模式、热水器预热，且支持周末禁用、雨天调整等例外条件。

2.3.2 规则配置与执行流程

底层规则 SQL 示例：device_type='temperature_sensor' AND room_id='master_bedroom' AND hour=7 AND day_of_week BETWEEN 1 AND 5

2.3.3 落地效果

2.4 生产级优化：解决'规则匹配延迟飙升'问题

2.4.1 问题爆发场景

当小区用户规则总数突破 10 万条时，Flink Task 的规则匹配耗时从 12ms/条飙升至 86ms/条。

2.4.2 根因定位

1. 遍历效率低下：每条设备状态需遍历所有 10 万条规则。
2. SQL 重复解析：相同规则的条件 SQL 被重复解析为 AST。
3. 状态存储无序：广播状态中的规则以 ruleId 为 key 无序存储。

2.4.3 优化方案落地

1. 规则二级索引优化：一级 key=userId+roomId，二级 key=ruleId。
2. SQL 预解析缓存：新增 ConcurrentHashMap 缓存 SqlNode。
3. 规则优先级排序：高频场景优先匹配。

2.4.4 优化前后对比

三、核心场景 2：场景化节能优化 —— 从'被动节能'到'预判调度'

3.1 行业痛点

传统节能模式多为'一刀切'，导致体验差。68% 的业主有'出门忘关设备'经历，设备空转浪费明显。

3.2 解决方案：'预测 - 调度 - 反馈'节能闭环

通过 ARIMA 模型预测未来 24 小时能耗需求，再用贪心算法生成错峰用电调度计划。

3.2.1 节能架构核心流程

[图片：节能架构流程图]

3.2.2 核心数据模型

3.2.2.1 能耗数据实体类

package com.smarthome.entity;
import lombok.Data;
import java.io.Serializable;

@Data
public class EnergyConsumption implements Serializable {
    private String deviceId;
    private String deviceType;
    private float energyKwh;
    private int runDuration;
    private long startTime;
    private long endTime;
    private String weather;
    private float outdoorTemp;
}

3.2.2.2 节能调度计划实体类

package com.smarthome.entity;
import lombok.Data;
import java.io.Serializable;

@Data
public class EnergySchedule implements Serializable {
    private Long scheduleId;
    private String deviceId;
    private int startHour;
    private int endHour;
    private String actionJson;
    private float energyForecast;
    private String priceType;
}

3.2.3 关键工具类：WeatherUtil

package com.smarthome.util;
import com.alibaba.fastjson.JSONObject;

public class WeatherUtil {
    public static JSONObject getCityWeather(String cityAdcode) {
        // 调用高德天气 API，带缓存逻辑
        return null;
    }

    public static float getWeatherFactor(String weather, float outdoorTemp) {
        float factor = 1.0f;
        if ("rain".equals(weather)) factor += 0.2f;
        if (outdoorTemp > 35) factor += 0.15f;
        return Math.max(0.8f, Math.min(1.5f, factor));
    }
}

3.2.4 核心算法实现：ARIMA 能耗预测

package com.smarthome.algorithm;
import org.apache.commons.math3.stat.regression.OLSMultipleLinearRegression;

public class ArimaEnergyPredictor {
    private static final int P = 2;
    private static final int D = 1;
    private static final int Q = 2;

    public double[] predictHourlyEnergy(String userId, String deviceId, String cityAdcode) {
        // 1. 拉取历史数据
        // 2. 异常值过滤（3σ原则）
        // 3. 差分去趋势
        // 4. 训练 AR 模型（最小二乘估计）
        // 5. 训练 MA 模型（极大似然估计）
        // 6. 预测并逆差分还原
        return new double[24];
    }

    public int[] generateEnergySchedule(String userId, String deviceId, String deviceType, double[] predictEnergy, String cityAdcode) {
        // 贪心算法：优先谷电时段运行
        return new int[24];
    }
}

3.2.5 节能调度执行 Job

package com.smarthome.flink.job;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class EnergyScheduleExecuteJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(300000);
        env.setParallelism(8);

        DataStream<EnergySchedule> scheduleStream = KafkaSourceBuilder.build(env, "energy_schedule_topic", "group", new SimpleStringSchema());
        DataStream<String> controlStream = scheduleStream.process(new ProcessFunction<>());
        controlStream.addSink(new DeviceControlSink("ssl://mqtt-broker:8883"));

        env.execute("Energy Schedule Execute Job");
    }
}

3.3 真实案例：上海仁恒河滨城'全屋家电错峰调度'

3.3.1 需求背景

用户希望热水器在谷电时段加热，空调在峰电时段调高温度，洗衣机自动在便宜时段运行。

3.3.2 落地方案与执行细节

3.3.3 落地效果

3.4 生产级优化：解决