scren_job定时任务城市臭氧滑动值计算逻辑修改

jinpengyong

2022-01-20 0aca12e52beb725c489a83a8624f00af355aa5fa

 screen-job/src/main/java/com/moral/api/service/impl/CityAqiServiceImpl.java

@@ -23,11 +23,13 @@
import org.springframework.util.ObjectUtils;
import org.springframework.web.client.RestTemplate;

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;

/**
@@ -61,7 +63,7 @@
        //存入数据库是time字段实际时间
        Date dataTime = DateUtils.addHours(time, -1);
        Date start = null;
        if (DateUtils.getHour(time) >= 8) {
        if (DateUtils.getHour(time) >= 8 || DateUtils.getHour(time) == 0) {
            start = DateUtils.addHours(time, -8);
        }

@@ -70,7 +72,28 @@
        wrapper.select("city_code", "city_id").eq("is_delete", Constants.NOT_DELETE);
        List<CityConfigAqi> list = cityConfigAqiService.list(wrapper);

        //获取历史数据，用于计算臭氧滑动值
        Map<Integer, List<Map<String, Object>>> cityData = null;
        if (start != null) {
            QueryWrapper<CityAqi> queryWrapper = new QueryWrapper<>();
            List<Integer> cityCodes = list.stream().map(CityConfigAqi::getCityCode)
                    .collect(Collectors.toList());

            queryWrapper.select("city_code", "time", "value")
                    .ge("time", DateUtils.dateToDateString(start))
                    .in("city_code", cityCodes);
            List<Map<String, Object>> selectMaps = cityAqiMapper.selectMaps(queryWrapper);

            //按city_code分组
            cityData = selectMaps.stream()
                    .collect(Collectors.groupingBy(o -> Integer.parseInt(o.get("city_code").toString())));
        }


        List<CityAqi> cityAqiList = new ArrayList<>();

        for (CityConfigAqi cityConfigAqi : list) {
            CityAqi cityAqi = new CityAqi();
            Integer cityCode = cityConfigAqi.getCityCode();
            Map<String, Object> data;
            try {
@@ -101,18 +124,15 @@
            aqi.put("AQI", aqi.remove("value"));

            //o3滑动值计算
            if (start != null) {
                QueryWrapper<CityAqi> queryWrapper = new QueryWrapper<>();
                queryWrapper.select("time", "value")
                        .eq("city_code", cityCode)
                        .ge("time", DateUtils.dateToDateString(start));
                List<Map<String, Object>> cityAqis = cityAqiMapper.selectMaps(queryWrapper);
            if (cityData != null) {
                List<Map<String, Object>> cityAqis = cityData.get(cityCode);

                Map<String, Object> params = new HashMap<>();
                params.put("time", DateUtils.dateToDateString(dataTime, DateUtils.yyyy_MM_dd_HH_mm_ss_S_EN));
                params.put("value", JSONObject.toJSONString(aqi));
                cityAqis.add(params);
                if (cityAqis.size() >= 6) {
                    OptionalDouble average = cityAqis.parallelStream().flatMapToDouble(v -> {
                    OptionalDouble average = cityAqis.stream().flatMapToDouble(v -> {
                        Map<String, Object> dataValue = JSONObject.parseObject((String) v.get("value"), Map.class);
                        double o3 = Double.parseDouble(dataValue.get("O3").toString());
                        return DoubleStream.of(o3);
@@ -142,14 +162,14 @@
            aqi.put("primaryPollutant", aqiAndPollutant.getPrimaryPollutantNames());


            CityAqi cityAqi = new CityAqi();
            cityAqi.setCityCode(cityCode);
            cityAqi.setTime(dataTime);
            cityAqi.setValue(JSONObject.toJSONString(aqi));
            cityAqiMapper.insert(cityAqi);
            cityAqiList.add(cityAqi);
            //存入redis
            redisTemplate.opsForHash().put(RedisConstants.CITY_AQI, cityCode.toString(), aqi);
        }
        cityAqiMapper.insertCityAqi(cityAqiList);
    }

    @Override