package com.moral.api.service.impl; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.util.ObjectUtils; import java.util.ArrayList; import java.util.Date; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.OptionalDouble; import java.util.stream.Collectors; import java.util.stream.DoubleStream; import com.alibaba.fastjson.JSONObject; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.moral.api.entity.Sensor; import com.moral.api.mapper.HistoryFiveMinutelyMapper; import com.moral.api.service.HistoryFiveMinutelyService; import com.moral.api.service.HistoryMinutelyService; import com.moral.api.service.SensorService; import com.moral.constant.Constants; import com.moral.util.DateUtils; @Service public class HistoryFiveMinutelyServiceImpl implements HistoryFiveMinutelyService { @Autowired private HistoryFiveMinutelyMapper historyFiveMinutelyMapper; @Autowired private HistoryMinutelyService historyMinutelyService; @Autowired private SensorService sensorService; @Override public void createTable(String timeUnits) { historyFiveMinutelyMapper.createTable(timeUnits); } @Override public void insertHistoryFiveMinutely() { //时间格式化:yyyy-MM-dd HH:mm String format = DateUtils.yyyy_MM_dd_HH_mm_EN; Date now = new Date(); Map params = new HashMap<>(); //开始时间 Date start = DateUtils.dataToTimeStampTime(DateUtils.getDateOfMin(now, -5), format); //结束时间 Date end = DateUtils.dataToTimeStampTime(now, format); params.put("start", start); params.put("end", end); //获取数据的分钟表后缀 String timeUnits = DateUtils.dateToDateString(start, DateUtils.yyyyMM_EN); params.put("timeUnits", timeUnits); //因子 QueryWrapper queryWrapper = new QueryWrapper<>(); queryWrapper.select("code").eq("is_delete", Constants.NOT_DELETE); List sensorCodes = sensorService.listObjs(queryWrapper); //获取所有设备的5分钟数据 List> fiveMinutelyData = historyMinutelyService.getHistoryFiveMinutelyData(params); //按mac分组 Map>> data = fiveMinutelyData.parallelStream() .collect(Collectors.groupingBy(o -> (String) o.get("mac"))); //存入数据库的结果集 List> insertData = new ArrayList<>(); data.forEach((key, value) -> { Map dataMap = new HashMap<>(); Map jsonMap = new HashMap<>(); dataMap.put("mac", key); dataMap.put("time", end); sensorCodes.forEach(sensorCode -> { OptionalDouble optionalDouble = value.parallelStream() .flatMapToDouble(v -> { Map dataValue = JSONObject.parseObject((String) v.get("value"), Map.class); Object sensorValue = dataValue.get(sensorCode.toString()); if (ObjectUtils.isEmpty(sensorValue)) { return null; } return DoubleStream.of(Double.parseDouble(sensorValue.toString())); }).average(); if (optionalDouble.isPresent()) { jsonMap.put(sensorCode.toString(), Double.parseDouble(String.format("%.4f", optionalDouble.getAsDouble()))); } }); dataMap.put("value", JSONObject.toJSONString(jsonMap)); insertData.add(dataMap); }); //5分钟表后缀 String insertTimeUnits = DateUtils.dateToDateString(now, DateUtils.yyyyMM_EN); historyFiveMinutelyMapper.insertHistoryFiveMinutely(insertData, insertTimeUnits); } public static void main(String[] args) { //模拟数据 List> list = new ArrayList<>(); Map map1 = new HashMap<>(); map1.put("mac", "p5dnd1234567"); map1.put("value", "{\"a0001\": 10, \"a0002\": 8, \"a0001-Flag\": \"N\", \"a0002-Flag\": \"N\"}"); Map map2 = new HashMap<>(); map2.put("mac", "p5dnd123456789"); map2.put("value", "{\"a0001\": 12, \"a0002\": 12, \"a0001-Flag\": \"N\", \"a0002-Flag\": \"N\"}"); Map map3 = new HashMap<>(); map3.put("mac", "p5dnd1234567"); map3.put("value", "{\"a0001\": 6, \"a0002\": 20, \"a0001-Flag\": \"N\", \"a0002-Flag\": \"N\", \"a0003\": 14}"); Map map4 = new HashMap<>(); map4.put("mac", "p5dnd1234567"); map4.put("value", "{\"a0001\": 4, \"a0002\": 16, \"a0001-Flag\": \"N\", \"a0002-Flag\": \"N\", \"a0003\": 16}"); list.add(map1); list.add(map2); list.add(map3); list.add(map4); //数据按mac分组 Map>> data = list.parallelStream().collect(Collectors.groupingBy(o -> (String) o.get("mac"))); System.out.println(data); //所有因子 List sensors = new ArrayList<>(); sensors.add("a0001"); sensors.add("a0002"); sensors.add("a0003"); //插入数据库结果集 List> insertData = new ArrayList<>(); data.forEach((key, value) -> { Map dataMap = new HashMap<>(); Map jsonMap = new HashMap<>(); dataMap.put("mac", key); sensors.forEach(sensorCode -> { OptionalDouble optionalDouble = value.parallelStream() .flatMapToDouble(v -> { Map dataValue = JSONObject.parseObject((String) v.get("value"), Map.class); Object sensorValue = dataValue.get(sensorCode.toString()); if (ObjectUtils.isEmpty(sensorValue)) { return null; } return DoubleStream.of(Double.parseDouble(sensorValue.toString())); }).average(); if (optionalDouble.isPresent()) { jsonMap.put(sensorCode.toString(), Double.parseDouble(String.format("%.4f", optionalDouble.getAsDouble()))); } }); dataMap.put("value", JSONObject.toJSONString(jsonMap)); insertData.add(dataMap); }); System.out.println("666==" + insertData); } }