Java函数式编程在物联网数据处理中的轻量化与效率

// 从 Kafka Topic 读取数据
Flux<SensorData> sensorData = KafkaFlux.create(KafkaProperties.builder()...build());

// 过滤出温度高于阈值的传感器数据
Flux<SensorData> filteredData = sensorData.filter(data -> data.getTemperature() > 50);

// 按设备分组，聚合每个设备的平均温度
Mono<Map<String, Double>> averageTemperatures = filteredData
    .groupBy(SensorData::getDeviceId)
    .reduce(new HashMap<>(), (map, data) -> {
        String deviceId = data.getDeviceId();
        Double temperature = data.getTemperature();
        Double currentAvg = map.getOrDefault(deviceId, 0.0);
        map.put(deviceId, (currentAvg + temperature) / 2);
        return map;
    });

// 将聚合结果保存到 MongoDB
averageTemperatures.flatMap(map -> Mono.fromCallable(() -> {
    MongoClient client = new MongoClient();
    MongoCollection<Document> collection = client.getDatabase("iot").getCollection("temperature");
    for (Map.Entry<String, Double> entry : map.entrySet()) {
        Document document = new Document().
            append("deviceId", entry.getKey()).
            append("temperature", entry.getValue());
        collection.insertOne(document);
    }
    client.close();
})).subscribe();