定时任务，kafka

jinpengyong

2021-06-23 69d00365e9f099b26c4fc7a298cabeb131956d8a

定时任务，kafka

 screen-api/src/main/resources/application-dev.yml

@@ -13,6 +13,8 @@
spring:
  profiles:
    active: dev
    include:
      moduleFormColumn
  application:
    name: screen-manage
  redis:
@@ -90,14 +92,11 @@
    enable:
      auto:
        commit: false
    group:
      id: test
    servers: 192.168.0.191:9092
    session:
      timeout: 6000
    topic: test_topic
    zookeeper:
      connect: 192.168.0.16:2181,192.168.0.17:2181,192.168.0.18:2181
      connect: 172.16.44.65:2181,172.16.44.67:2181,172.16.44.66:2181
  producer:
    batch:
      size: 4096
@@ -105,7 +104,7 @@
      memory: 40960
    linger: 1
    retries: 0
    servers: 192.168.0.16:9092,192.168.0.17:9092,192.168.0.18:9092
    servers: 72.16.44.65:9092,172.16.44.67:9092,172.16.44.66:9092
mvc:
  interceptor:
    exclude:

 screen-api/src/main/resources/mapper/GroupMapper.xml

@@ -17,7 +17,7 @@
        <result column="group_name" property="groupName"/>
    </resultMap>

    <select id="selectUserGroup" resultMap="GroupResultMap">
    <select id="selectUserGroup" resultMap="BaseResultMap">
        SELECT g.id,g.group_name
        FROM `group` g,`user_group` ug
        WHERE ug.user_id = #{userId}

 screen-api/src/main/resources/mapper/GroupMenuMapper.xml

@@ -2,16 +2,16 @@
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.GroupMenuMapper">

        <!-- 通用查询映射结果 -->
        <resultMap id="BaseResultMap" type="com.moral.api.entity.GroupMenu">
                    <id column="id" property="id" />
                    <result column="group_id" property="groupId" />
                    <result column="menu_id" property="menuId" />
                    <result column="channel_key" property="channelKey" />
                    <result column="organization_id" property="organizationId" />
                    <result column="create_time" property="createTime" />
                    <result column="update_time" property="updateTime" />
                    <result column="is_delete" property="isDelete" />
        </resultMap>
    <!-- 通用查询映射结果 -->
    <resultMap id="BaseResultMap" type="com.moral.api.entity.GroupMenu">
        <id column="id" property="id"/>
        <result column="group_id" property="groupId"/>
        <result column="menu_id" property="menuId"/>
        <result column="channel_key" property="channelKey"/>
        <result column="organization_id" property="organizationId"/>
        <result column="create_time" property="createTime"/>
        <result column="update_time" property="updateTime"/>
        <result column="is_delete" property="isDelete"/>
    </resultMap>

</mapper>

 screen-api/src/main/resources/mapper/MenuMapper.xml

@@ -20,7 +20,7 @@
    </sql>

    <select id="selectUserMenu" resultMap="BaseResultMap">
        select
        SELECT
        <include refid="Base_Column_List"/>
        FROM `user_group` ug,`group_menu` gm,`menu` m
        WHERE ug.user_id = #{userId}
@@ -33,14 +33,13 @@
    </select>

    <select id="selectOrganizationMenu" resultMap="BaseResultMap">
        select
        <include refid="Base_Column_List"/>
        FROM `menu` m, `organization_menu` om
        WHERE om.organization_id = #{orgId}
        AND m.id = om.menu_id
        AND om.channel_key = 0
        AND om.menu_id = m.id
        AND om.is_delete = 0
        SELECT
        DISTINCT m.id, m.name, m.url, m.icon, m.parent_id, m.order, m.create_time, m.update_time
        FROM `menu` m, `group_menu` gm
        WHERE gm.organization_id = #{orgId}
        AND m.id = gm.menu_id
        AND gm.channel_key = 0
        AND gm.is_delete = 0
        AND m.is_delete = 0
    </select>
</mapper>

 screen-common/src/main/java/com/moral/constant/Constants.java

@@ -120,5 +120,10 @@
     * */
    public static final String UN_ADJUST = "unadjust";

    /*
     * 离线设备状态码
     * */
    public static final String DEVICE_STATE_OFFLINE = "4";


}

 screen-common/src/main/java/com/moral/constant/KafkaConstants.java

@@ -5,15 +5,25 @@
    /**
     * 分钟数据主题
     */
    public static final String TOPIC_MINUTE = "test_topic";
    public static final String TOPIC_MINUTE = "minute";

    /**
     * 小时数据主题
     */
    public static final String TOPIC_HOUR = "hour";

    /*
     * 秒数据主题
     * */
    public static final String TOPIC_SECOND = "test";

    /**
     * 消费组
     * 用于将数据存入数据库的消费组
     */
    public static final String GROUP_ID = "test";
    public static final String GROUP_ID_INSERT = "insert";

    /**
     * 用于判断设备状态消费组
     */
    public static final String GROUP_ID_STATE = "state";
}

 screen-common/src/main/java/com/moral/constant/RedisConstants.java

@@ -47,4 +47,15 @@
    * 接收类型：Map<String,Device>
    * */
    public static final String DEVICE_INFO = "device_alarm_info";

    /*
     * 设备校准公式前缀
     * */
    public static final String ADJUST = "adjust";

    /*
     * 设备实时数据
     * */
    public static final String DEVICE_DATA = "data";

}

 screen-common/src/main/java/com/moral/util/DateUtils.java

@@ -67,6 +67,14 @@
     * 日期格式(HH:mm:ss)
     */
    public static final String HH_mm_ss_EN = "HH:mm:ss";
    /*
     * 日期格式(yyyy-MM-dd HH:mm)
     * */
    public static final String yyyy_MM_dd_HH_mm_EN = "yyyy-MM-dd HH:mm";
    /*
     * 日期格式(yyyy-MM-dd HH)
     * */
    public static final String yyyy_MM_dd_HH_EN = "yyyy-MM-dd HH";
    /**
     * DateFormat缓存
     */
@@ -443,14 +451,14 @@
    }

    /**
    * @Description: 获取指定日期day天后的日期
            * @Param: [date, day]
            * @return: java.util.Date
            * @Author: 陈凯裕
            * @Date: 2021/3/30
            */
     * @Description: 获取指定日期day天后的日期
     * @Param: [date, day]
     * @return: java.util.Date
     * @Author: 陈凯裕
     * @Date: 2021/3/30
     */
    public static Date getDateOfDay(Date date, int day) {
        if(date==null)
        if (date == null)
            return null;
        Calendar now = Calendar.getInstance(TimeZone.getDefault());
        now.setTime(date);
@@ -1293,11 +1301,23 @@
        return date;
    }

    //当前时间转换，只取到分钟
    public static Date convertDate(Date date) {
        String dateString = dateToDateString(date, yyyy_MM_dd_HH_mm_EN);
        return getDate(dateString, yyyy_MM_dd_HH_mm_EN);
    }

    //时间戳转换，只取时分秒
    public static Date dataToTimeStampTime(Date time, String dateFormat) {
        String dateString = dateToDateString(time, dateFormat);
        try {
            return getDateFormat(dateFormat).parse(dateString);
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Date date1 = new Date();
        Thread.sleep(1000);
        Date date2 = new Date();
        System.out.println(DateUtils.compareDateStr(date1, date2));
        System.out.println(new Date().getTime());
    }
}

 screen-job/src/main/java/com/moral/api/config/redis/RedisConfig.java

New file
@@ -0,0 +1,67 @@
package com.moral.api.config.redis;

import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;


@Configuration
public class RedisConfig {

    @Bean(name = "redisTemplate")
    @ConditionalOnMissingBean(StringRedisTemplate.class)  //此注解的作用  如果容器中没有RedisTemplate 那就注入    有就不注入了
    public RedisTemplate<String, Object> stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
        redisTemplate.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        // valuevalue采用jackson序列化方式
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value采用jackson序列化方式
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    public RedisTemplate createRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);

        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        // key采用String的序列化方式
        redisTemplate.setKeySerializer(stringRedisSerializer);
        // hash的key也采用String的序列化方式
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        // valuevalue采用jackson序列化方式
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        // hash的value采用jackson序列化方式
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}

 screen-job/src/main/java/com/moral/api/config/xxl/XxlJobConfig.java

File was deleted

 screen-job/src/main/java/com/moral/api/config/xxljob/XxlJobConfig.java

New file
@@ -0,0 +1,68 @@
package com.moral.api.config.xxljob;


import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;

@Configuration
@ComponentScan(basePackages = "com.moral.api.job")
@Slf4j
public class XxlJobConfig {

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.executor.appname}")
    private String appName;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private Integer port;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private Integer logRetentionDays;

    @Bean
    public XxlJobSpringExecutor xxlJobSpringExecutor() {
        log.info("xxl jon config init");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appName);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
        return xxlJobSpringExecutor;
    }

    /**
     * 针对多网卡、容器内部署等情况，可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP；
     *
     *      1、引入依赖：
     *          <dependency>
     *             <groupId>org.springframework.cloud</groupId>
     *             <artifactId>spring-cloud-commons</artifactId>
     *             <version>${version}</version>
     *         </dependency>
     *
     *      2、配置文件，或者容器启动变量
     *          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
     *
     *      3、获取IP
     *          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
     */
}

 screen-job/src/main/java/com/moral/api/entity/CityAqiConfig.java

New file
@@ -0,0 +1,50 @@
package com.moral.api.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.extension.activerecord.Model;
import com.baomidou.mybatisplus.annotation.TableId;
import java.io.Serializable;
import lombok.Data;
import lombok.EqualsAndHashCode;

/**
 * <p>
 * 墨迹天气aqi城市配置
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
@Data
@EqualsAndHashCode(callSuper = false)
public class CityAqiConfig extends Model<CityAqiConfig> {

    private static final long serialVersionUID = 1L;

    /**
     * 主键id
     */
    @TableId(value = "id", type = IdType.AUTO)
    private Integer id;

    /**
     * 城市code
     */
    private String cityCode;

    /**
     * 城市名字
     */
    private String cityName;

    /**
     * 墨迹天气接口中城市id
     */
    private Integer cityId;

    @Override
    protected Serializable pkVal() {
        return this.id;
    }

}

 screen-job/src/main/java/com/moral/api/entity/Device.java

New file
@@ -0,0 +1,132 @@
package com.moral.api.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.extension.activerecord.Model;
import com.baomidou.mybatisplus.annotation.TableId;
import java.time.LocalDateTime;
import java.io.Serializable;
import lombok.Data;
import lombok.EqualsAndHashCode;

/**
 * <p>
 * 设备表
 * </p>
 *
 * @author moral
 * @since 2021-06-22
 */
@Data
@EqualsAndHashCode(callSuper = false)
public class Device extends Model<Device> {

    private static final long serialVersionUID = 1L;

    /**
     * 主键id
     */
    @TableId(value = "id", type = IdType.AUTO)
    private Integer id;

    /**
     * 设备名称
     */
    private String name;

    /**
     * mac号
     */
    private String mac;

    /**
     * 设备地址
     */
    private String address;

    /**
     * 经度
     */
    private Double longitude;

    /**
     * 纬度
     */
    private Double latitude;

    /**
     * 设备状态，与字典表关联
     */
    private String state;

    /**
     * 维护人id，多个逗号隔开，来源于manage_account
     */
    private String operateIds;

    /**
     * 站点id
     */
    private Integer monitorPointId;

    /**
     * 组织id
     */
    private Integer organizationId;

    /**
     * 设备型号id
     */
    private Integer deviceVersionId;

    /**
     * 行业，与字典表关联
     */
    private String profession;

    /**
     * 设备工艺，1：烟道；2：厂界；3：车间，基本数据在字典表
     */
    private String tech;

    /**
     * 设备检测器，与字典表关联
     */
    private String detector;

    /**
     * 采购商，与字典表关联
     */
    private String purchaser;

    /**
     * 创建(生产)时间
     */
    private LocalDateTime createTime;

    /**
     * 更新时间
     */
    private LocalDateTime updateTime;

    /**
     * 安装时间
     */
    private LocalDateTime installTime;

    /**
     * 逻辑删除,0：不删除；1：删除
     */
    private String isDelete;

    /**
     * 设备工艺扩展字段
     */
    private String extend;


    @Override
    protected Serializable pkVal() {
        return this.id;
    }

}

 screen-job/src/main/java/com/moral/api/entity/HistoryAqi.java

New file
@@ -0,0 +1,45 @@
package com.moral.api.entity;

import com.baomidou.mybatisplus.extension.activerecord.Model;
import java.io.Serializable;
import java.util.Date;

import lombok.Data;
import lombok.EqualsAndHashCode;

/**
 * <p>
 * 国控aqi数据表
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
@Data
@EqualsAndHashCode(callSuper = false)
public class HistoryAqi extends Model<HistoryAqi> {

    private static final long serialVersionUID = 1L;

    /**
     * 城市code
     */
    private String cityCode;

    /**
     * 时间
     */
    private Date time;

    /**
     * 数据
     */
    private String value;


    @Override
    protected Serializable pkVal() {
        return null;
    }

}

 screen-job/src/main/java/com/moral/api/entity/HistoryDaily.java

New file
@@ -0,0 +1,36 @@
package com.moral.api.entity;

import com.baomidou.mybatisplus.extension.activerecord.Model;
import java.io.Serializable;
import java.util.Date;

import lombok.Data;
import lombok.EqualsAndHashCode;

/**
 * <p>
 *
 * </p>
 *
 * @author moral
 * @since 2021-06-11
 */
@Data
@EqualsAndHashCode(callSuper = false)
public class HistoryDaily extends Model<HistoryDaily> {

    private static final long serialVersionUID = 1L;

    private String mac;

    private Date time;

    private String value;


    @Override
    protected Serializable pkVal() {
        return null;
    }

}

 screen-job/src/main/java/com/moral/api/mapper/CityAqiConfigMapper.java

New file
@@ -0,0 +1,16 @@
package com.moral.api.mapper;

import com.moral.api.entity.CityAqiConfig;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;

/**
 * <p>
 * 墨迹天气aqi城市配置 Mapper 接口
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
public interface CityAqiConfigMapper extends BaseMapper<CityAqiConfig> {

}

 screen-job/src/main/java/com/moral/api/mapper/DeviceMapper.java

New file
@@ -0,0 +1,16 @@
package com.moral.api.mapper;

import com.moral.api.entity.Device;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;

/**
 * <p>
 * 设备表 Mapper 接口
 * </p>
 *
 * @author moral
 * @since 2021-06-22
 */
public interface DeviceMapper extends BaseMapper<Device> {

}

 screen-job/src/main/java/com/moral/api/mapper/HistoryAqiMapper.java

New file
@@ -0,0 +1,17 @@
package com.moral.api.mapper;

import com.moral.api.entity.HistoryAqi;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;

/**
 * <p>
 * 国控aqi数据表 Mapper 接口
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
public interface HistoryAqiMapper extends BaseMapper<HistoryAqi> {

}


 screen-job/src/main/java/com/moral/api/mapper/HistoryDailyMapper.java

New file
@@ -0,0 +1,21 @@
package com.moral.api.mapper;

import java.util.List;
import java.util.Map;

import com.moral.api.entity.HistoryDaily;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;

/**
 * <p>
 *  Mapper 接口
 * </p>
 *
 * @author moral
 * @since 2021-06-11
 */
public interface HistoryDailyMapper extends BaseMapper<HistoryDaily> {

    void insertHistoryDaily(List<HistoryDaily> list);

}

 screen-job/src/main/java/com/moral/api/mapper/HistoryFiveMinutelyMapper.java

New file
@@ -0,0 +1,15 @@
package com.moral.api.mapper;


import org.apache.ibatis.annotations.Param;

import java.util.List;
import java.util.Map;

public interface HistoryFiveMinutelyMapper {

    void createTable(String timeUnits);

    void insertHistoryFiveMinutely(@Param("list") List<Map<String, Object>> list, @Param("timeUnits") String timeUnits);

}

 screen-job/src/main/java/com/moral/api/mapper/HistoryMinutelyMapper.java

New file
@@ -0,0 +1,11 @@
package com.moral.api.mapper;

import java.util.List;
import java.util.Map;

public interface HistoryMinutelyMapper {

    void createTable(String timeUnits);

    List<Map<String, Object>> getHistoryFiveMinutelyData(Map<String,Object> params);
}

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

New file
@@ -0,0 +1,20 @@
package com.moral.api.service;

import java.util.List;

import com.moral.api.entity.CityAqiConfig;
import com.baomidou.mybatisplus.extension.service.IService;

/**
 * <p>
 * 墨迹天气aqi城市配置 服务类
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
public interface CityAqiConfigService extends IService<CityAqiConfig> {

    List<CityAqiConfig> getCityAqiConfigs();

}

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

New file
@@ -0,0 +1,20 @@
package com.moral.api.service;

import com.moral.api.entity.Device;
import com.baomidou.mybatisplus.extension.service.IService;

/**
 * <p>
 * 设备表 服务类
 * </p>
 *
 * @author moral
 * @since 2021-06-22
 */
public interface DeviceService extends IService<Device> {

    void judgeOffLineDevice();

    void updateDeviceState(Device device);

}

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

New file
@@ -0,0 +1,10 @@
package com.moral.api.service;

import com.baomidou.mybatisplus.extension.service.IService;
import com.moral.api.entity.HistoryAqi;

public interface HistoryAqiService extends IService<HistoryAqi> {

    void insertHistoryAqi();

}

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

New file
@@ -0,0 +1,21 @@
package com.moral.api.service;

import java.util.List;
import java.util.Map;

import com.moral.api.entity.HistoryDaily;
import com.baomidou.mybatisplus.extension.service.IService;

/**
 * <p>
 * 服务类
 * </p>
 *
 * @author moral
 * @since 2021-06-11
 */
public interface HistoryDailyService extends IService<HistoryDaily> {

    void insertHistoryDaily(List<HistoryDaily> list);

}

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

New file
@@ -0,0 +1,9 @@
package com.moral.api.service;

public interface HistoryFiveMinutelyService {

    void createTable(String timeUnits);

    void insertHistoryFiveMinutely();

}

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

New file
@@ -0,0 +1,12 @@
package com.moral.api.service;

import java.util.List;
import java.util.Map;

public interface HistoryMinutelyService {

    void createTable(String timeUnits);

    List<Map<String, Object>> getHistoryFiveMinutelyData(Map<String, Object> params);

}

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

File was deleted

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

New file
@@ -0,0 +1,31 @@
package com.moral.api.service.impl;

import com.moral.api.entity.CityAqiConfig;
import com.moral.api.mapper.CityAqiConfigMapper;
import com.moral.api.service.CityAqiConfigService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * <p>
 * 墨迹天气aqi城市配置 服务实现类
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
@Service
public class CityAqiConfigServiceImpl extends ServiceImpl<CityAqiConfigMapper, CityAqiConfig> implements CityAqiConfigService {

    @Autowired
    private CityAqiConfigMapper cityAqiConfigMapper;

    @Override
    public List<CityAqiConfig> getCityAqiConfigs() {
        return cityAqiConfigMapper.selectList(null);
    }
}

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

New file
@@ -0,0 +1,69 @@
package com.moral.api.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.moral.api.entity.Device;
import com.moral.api.mapper.DeviceMapper;
import com.moral.api.service.DeviceService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.moral.constant.Constants;
import com.moral.constant.RedisConstants;
import com.moral.util.DateUtils;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.List;
import java.util.Map;

/**
 * <p>
 * 设备表 服务实现类
 * </p>
 *
 * @author moral
 * @since 2021-06-22
 */
@Service
public class DeviceServiceImpl extends ServiceImpl<DeviceMapper, Device> implements DeviceService {

    @Autowired
    private DeviceMapper deviceMapper;

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void judgeOffLineDevice() {
        QueryWrapper<Device> queryWrapper = new QueryWrapper<>();
        queryWrapper.ne("state", Constants.DEVICE_STATE_OFFLINE);
        //获取所有在线设备
        List<Device> devices = deviceMapper.selectList(queryWrapper);
        for (Device device : devices) {
            Map<String, Object> data = getDataFromRedis(device.getMac());
            device.setState(Constants.DEVICE_STATE_OFFLINE);
            if (data != null && data.containsKey("DataTime")) {
                long time = Long.parseLong(data.get("DataTime").toString());
                //超过两分钟无数据就离线
                if (DateUtils.getDateOfMin(new Date(time), 2).getTime() < new Date().getTime()) {
                    updateDeviceState(device);
                }
            } else {
                updateDeviceState(device);
            }
        }
    }

    @Override
    public void updateDeviceState(Device device) {
        UpdateWrapper<Device> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id", device.getId()).set("state", device.getState());
        deviceMapper.update(null, updateWrapper);
    }

    private Map<String, Object> getDataFromRedis(String mac) {
        return (Map<String, Object>) redisTemplate.opsForValue().get(RedisConstants.DEVICE_DATA + "_" + mac);
    }
}

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

New file
@@ -0,0 +1,77 @@
package com.moral.api.service.impl;

import com.alibaba.fastjson.JSON;
import com.moral.api.entity.CityAqiConfig;
import com.moral.api.entity.HistoryAqi;
import com.moral.api.mapper.HistoryAqiMapper;
import com.moral.api.service.CityAqiConfigService;
import com.moral.api.service.HistoryAqiService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.web.client.RestTemplate;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * <p>
 * 国控aqi数据表 服务实现类
 * </p>
 *
 * @author moral
 * @since 2021-06-15
 */
@Service
@Slf4j
public class HistoryAqiServiceImpl extends ServiceImpl<HistoryAqiMapper, HistoryAqi> implements HistoryAqiService {

    @Autowired
    private HistoryAqiMapper historyAqiMapper;

    @Autowired
    private CityAqiConfigService cityAqiConfigService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public void insertHistoryAqi() {
        RestTemplate restTemplate = new RestTemplate();
        //获取aqi城市配置
        List<CityAqiConfig> cityAqiConfigs = cityAqiConfigService.getCityAqiConfigs();
        for (CityAqiConfig cityAqiConfig : cityAqiConfigs) {
            Map<String, Object> mjMap = restTemplate.getForObject("http://sapi.7drlb.com/api/mj?cityID={1}&apiKey=aqi", Map.class, cityAqiConfig.getCityId());
            if (ObjectUtils.isEmpty(mjMap)) {
                continue;
            }
            HistoryAqi historyAqi = new HistoryAqi();
            //city_code
            String cityCode = cityAqiConfig.getCityCode();
            historyAqi.setCityCode(cityCode);
            Map<String, Object> value = new HashMap<>();
            Map<String, Object> data = (Map<String, Object>) ((Map) mjMap.get("data")).get("aqi");
            //数据时间
            historyAqi.setTime(new Date(Long.parseLong(data.get("pubtime").toString())));
            value.put("PM25", data.get("pm25C"));
            value.put("PM10", data.get("pm10C"));
            value.put("SO2", data.get("so2C"));
            value.put("NO2", data.get("no2C"));
            value.put("CO", data.get("coC"));
            value.put("O3", data.get("o3C"));
            value.put("AQI", data.get("value"));
            //数据
            historyAqi.setValue(JSON.toJSONString(value));
            //数据存入数据库
            historyAqiMapper.insert(historyAqi);
            //存入redis
            redisTemplate.opsForHash().putAll("aqi_" + cityCode, value);
        }
    }
}

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

New file
@@ -0,0 +1,33 @@
package com.moral.api.service.impl;

import com.moral.api.entity.HistoryDaily;
import com.moral.api.mapper.HistoryDailyMapper;
import com.moral.api.service.HistoryDailyService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author moral
 * @since 2021-06-11
 */
@Service
public class HistoryDailyServiceImpl extends ServiceImpl<HistoryDailyMapper, HistoryDaily> implements HistoryDailyService {

    @Autowired
    private HistoryDailyMapper historyDailyMapper;

    @Override
    public void insertHistoryDaily(List<HistoryDaily> list) {
        System.out.println(list);
        historyDailyMapper.insertHistoryDaily(list);
    }
}

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

New file
@@ -0,0 +1,64 @@
package com.moral.api.service.impl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.moral.api.mapper.HistoryFiveMinutelyMapper;
import com.moral.api.service.HistoryFiveMinutelyService;
import com.moral.api.service.HistoryMinutelyService;
import com.moral.util.DateUtils;

@Service
public class HistoryFiveMinutelyServiceImpl implements HistoryFiveMinutelyService {

    @Autowired
    private HistoryFiveMinutelyMapper historyFiveMinutelyMapper;

    @Autowired
    private HistoryMinutelyService historyMinutelyService;

    @Override
    public void createTable(String timeUnits) {
        historyFiveMinutelyMapper.createTable(timeUnits);
    }

    @Override
    public void insertHistoryFiveMinutely() {
        Map<String, Object> params = new HashMap<>();
        //开始时间（分钟）
        Date start = DateUtils.convertDate(DateUtils.getDateOfMin(-5));
        //结束时间（分钟）
        Date end = DateUtils.convertDate(new Date());
        params.put("start", start);
        params.put("end", end);
        //分钟表后缀
        String timeUnits = DateUtils.dateToDateString(start, DateUtils.yyyyMM_EN);
        params.put("timeUnits", timeUnits);

        //因子

        params.put("sensorKeys", null);

        historyMinutelyService.getHistoryFiveMinutelyData(params);
        /*List<Map<String, Object>> list = new ArrayList<>();
        Map<String, Object> map = new HashMap<>();
        map.put("mac", "123456");
        Date date = new Date();
        map.put("time", date);
        Map<String, Object> value = new HashMap<>();
        value.put("e1", 10);
        value.put("e2", 20);
        map.put("value", JSON.toJSONString(value));
        list.add(map);*/

        historyFiveMinutelyMapper.insertHistoryFiveMinutely(null, null);
    }
}

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

New file
@@ -0,0 +1,27 @@
package com.moral.api.service.impl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

import com.moral.api.mapper.HistoryMinutelyMapper;
import com.moral.api.service.HistoryMinutelyService;

@Service
public class HistoryMinutelyServiceImpl implements HistoryMinutelyService {

    @Autowired
    private HistoryMinutelyMapper historyMinutelyMapper;

    @Override
    public void createTable(String timeUnits) {
        historyMinutelyMapper.createTable(timeUnits);
    }

    @Override
    public List<Map<String, Object>> getHistoryFiveMinutelyData(Map<String, Object> params) {
        return historyMinutelyMapper.getHistoryFiveMinutelyData(params);
    }
}

 screen-job/src/main/java/com/moral/api/task/CreateTableTask.java

New file
@@ -0,0 +1,52 @@
package com.moral.api.task;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.moral.api.service.HistoryFiveMinutelyService;
import com.moral.api.service.HistoryMinutelyService;
import com.moral.constant.Constants;
import com.moral.util.DateUtils;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;

@Component
public class CreateTableTask {

    @Autowired
    private HistoryFiveMinutelyService historyFiveMinutelyService;

    @Autowired
    private HistoryMinutelyService historyMinutelyService;


    //分钟表创建任务
    @XxlJob("createHistoryMinutelyTable")
    public ReturnT createHistoryMinutelyTable() {
        String timeUnits = DateUtils.getDateStringOfMon(1, DateUtils.yyyyMM_EN);
        try {
            //已校准分钟表
            historyMinutelyService.createTable(timeUnits);
            //未校准分钟表
            historyMinutelyService.createTable(timeUnits + "_" + Constants.UN_ADJUST);
        } catch (Exception e) {
            XxlJobHelper.log(e.getMessage());
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }

    //5分钟表创建任务
    @XxlJob("createHistoryFiveMinutelyTable")
    public ReturnT createHistoryFiveMinutelyTable() {
        String timeUnits = DateUtils.getDateStringOfMon(1, DateUtils.yyyyMM_EN);
        try {
            historyFiveMinutelyService.createTable(timeUnits);
        } catch (Exception e) {
            XxlJobHelper.log(e.getMessage());
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }
}

 screen-job/src/main/java/com/moral/api/task/HistoryAqiInsertTask.java

New file
@@ -0,0 +1,27 @@
package com.moral.api.task;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.moral.api.service.HistoryAqiService;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;

@Component
public class HistoryAqiInsertTask {

    @Autowired
    private HistoryAqiService historyAqiService;

    @XxlJob("insertHistoryAqi")
    public ReturnT insertHistoryAqi() {
        try {
            historyAqiService.insertHistoryAqi();
        } catch (Exception e) {
            XxlJobHelper.log(e.getMessage());
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }
}

 screen-job/src/main/java/com/moral/api/task/HistoryTableInsertTask.java

New file
@@ -0,0 +1,74 @@
package com.moral.api.task;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.moral.api.entity.HistoryDaily;
import com.moral.api.service.HistoryDailyService;
import com.moral.api.service.HistoryFiveMinutelyService;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;

@Component
public class HistoryTableInsertTask {

    @Autowired
    private HistoryDailyService historyDailyService;

    @Autowired
    private HistoryFiveMinutelyService historyFiveMinutelyService;

    //5分钟数据统计
    @XxlJob("insertHistoryFiveMinutely")
    public ReturnT insertHistoryFiveMinutely() {
        try {
           /* List<Map<String, Object>> list = new ArrayList<>();
            Map<String, Object> map = new HashMap<>();
            map.put("mac", "123456");
            Date date = new Date();
            map.put("time", date);
            Map<String, Object> value = new HashMap<>();
            value.put("e1", 10);            value.put("e2", 20);
            map.put("value", JSON.toJSONString(value));
            list.add(map);
            String timeUnits = DateUtils.dateToDateString(date, DateUtils.yyyyMM_EN);*/

            historyFiveMinutelyService.insertHistoryFiveMinutely();
        } catch (Exception e) {
            XxlJobHelper.log(e.getMessage());
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }

    //天数据统计
    @XxlJob("insertHistoryDaily")
    public ReturnT insertHistoryDaily() {
        try {
            List<HistoryDaily> list = new ArrayList<>();
            HistoryDaily historyDaily = new HistoryDaily();
            historyDaily.setMac("123456");
            historyDaily.setTime(new Date());
            Map<String, Object> value = new HashMap<>();
            value.put("e1", 1);
            value.put("e2", 2);
            historyDaily.setValue(JSON.toJSONString(value));
            for (int i = 0; i < 20000; i++) {
                list.add(historyDaily);
            }
            historyDailyService.insertHistoryDaily(list);
        } catch (Exception e) {
            XxlJobHelper.log(e.getMessage());
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }
}

 screen-job/src/main/java/com/moral/api/task/JudgeOffLineDeviceTask.java

New file
@@ -0,0 +1,28 @@
package com.moral.api.task;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.moral.api.service.DeviceService;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;

@Component
public class JudgeOffLineDeviceTask {

    @Autowired
    private DeviceService deviceService;

    //判断设备是否离线
    @XxlJob("judgeOffLineDevice")
    public ReturnT judgeOffLineDevice() {
        try {
            deviceService.judgeOffLineDevice();
        } catch (Exception e) {
            XxlJobHelper.log(e.getMessage());
            return ReturnT.FAIL;
        }
        return ReturnT.SUCCESS;
    }
}

 screen-job/src/main/resources/application-dev.yml

@@ -81,6 +81,7 @@

logging:
  config: classpath:logback.xml

xxl:
  job:
    admin:

 screen-job/src/main/resources/logback-spring.xml

New file
@@ -0,0 +1,56 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>

    <springProperty scope="context" name="springAppName" source="spring.application.name"/>
    <!-- 日志在工程中的输出位置 -->
    <property name="LOG_FILE" value="/home/moral/soft/log/${springAppName}"/>
    <!-- 控制台的日志输出样式 -->
    <property name="CONSOLE_LOG_PATTERN"
              value="%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}) [%X{logseq}]{faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr([${springAppName}]){yellow} %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}"/>

    <property name="FILE_LOG_PATTERN"
              value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%X{logseq}] [%thread] %X{T} [%level] %logger.%method:%line %msg%n"/>

    <!-- 控制台Appender -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
            <level>INFO</level>
        </filter>
        <encoder>
            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
            <charset>utf8</charset>
        </encoder>
    </appender>
    <!-- 日志记录Appender -->
    <appender name="screenJobLog" class="ch.qos.logback.classic.sift.SiftingAppender">
        <discriminator>
            <key>taskId</key>
            <defaultValue>default</defaultValue>
        </discriminator>
        <sift>
            <appender name="file" class="ch.qos.logback.core.rolling.RollingFileAppender">

                <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
                    <level>INFO</level>
                </filter>
                <file>${LOG_FILE}/%d{yyyy-MM-dd}/${springAppName}.log</file>
                <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
                    <maxFileSize>30MB</maxFileSize>
                    <fileNamePattern>${LOG_FILE}/%d{yyyy-MM-dd}/${springAppName}-%d{yyyy-MM-dd}.log%i.log</fileNamePattern>
                    <maxHistory>30</maxHistory>
                </rollingPolicy>
                <encoder>
                    <pattern>${FILE_LOG_PATTERN}</pattern>
                    <charset>utf8</charset>
                </encoder>
            </appender>
        </sift>
    </appender>


    <root level="INFO">
        <appender-ref ref="console"/>
        <appender-ref ref="screenJobLog"/>
    </root>
</configuration>

 screen-job/src/main/resources/mapper/CityAqiConfigMapper.xml

New file
@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.CityAqiConfigMapper">

    <!-- 通用查询映射结果 -->
    <resultMap id="BaseResultMap" type="com.moral.api.entity.CityAqiConfig">
        <id column="id" property="id"/>
        <result column="city_code" property="cityCode"/>
        <result column="city_name" property="cityName"/>
        <result column="city_id" property="cityId"/>
    </resultMap>

</mapper>

 screen-job/src/main/resources/mapper/DeviceMapper.xml

New file
@@ -0,0 +1,29 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.DeviceMapper">

        <!-- 通用查询映射结果 -->
        <resultMap id="BaseResultMap" type="com.moral.api.entity.Device">
                    <id column="id" property="id" />
                    <result column="name" property="name" />
                    <result column="mac" property="mac" />
                    <result column="address" property="address" />
                    <result column="longitude" property="longitude" />
                    <result column="latitude" property="latitude" />
                    <result column="state" property="state" />
                    <result column="operate_ids" property="operateIds" />
                    <result column="monitor_point_id" property="monitorPointId" />
                    <result column="organization_id" property="organizationId" />
                    <result column="device_version_id" property="deviceVersionId" />
                    <result column="profession" property="profession" />
                    <result column="tech" property="tech" />
                    <result column="detector" property="detector" />
                    <result column="purchaser" property="purchaser" />
                    <result column="create_time" property="createTime" />
                    <result column="update_time" property="updateTime" />
                    <result column="install_time" property="installTime" />
                    <result column="is_delete" property="isDelete" />
                    <result column="extend" property="extend" />
        </resultMap>

</mapper>

 screen-job/src/main/resources/mapper/HistoryAqiMapper.xml

New file
@@ -0,0 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.HistoryAqiMapper">

        <!-- 通用查询映射结果 -->
        <resultMap id="BaseResultMap" type="com.moral.api.entity.HistoryAqi">
                    <result column="city_code" property="cityCode" />
                    <result column="time" property="time" />
                    <result column="value" property="value" />
        </resultMap>

</mapper>

 screen-job/src/main/resources/mapper/HistoryDailyMapper.xml

New file
@@ -0,0 +1,19 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.HistoryDailyMapper">

    <!-- 通用查询映射结果 -->
    <resultMap id="BaseResultMap" type="com.moral.api.entity.HistoryDaily">
        <result column="mac" property="mac"/>
        <result column="time" property="time"/>
        <result column="value" property="value"/>
    </resultMap>

    <insert id="insertHistoryDaily" parameterType="com.moral.api.entity.HistoryDaily">
        INSERT INTO history_daily
        VALUES
        <foreach collection="list" item="item" separator=",">
            (#{item.mac},#{item.time},#{item.value})
        </foreach>
    </insert>
</mapper>

 screen-job/src/main/resources/mapper/HistoryFiveMinutelyMapper.xml

New file
@@ -0,0 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.HistoryFiveMinutelyMapper">

    <update id="createTable" parameterType="String">
        CREATE TABLE  IF NOT EXISTS `history_five_minutely_${timeUnits}` (
            `mac` VARCHAR (20) DEFAULT NULL COMMENT '设备mac',
            `time` datetime DEFAULT NULL COMMENT '数据时间',
            `value` json DEFAULT NULL COMMENT '数据',
            KEY `idx_mac` (`mac`),
            KEY `idx_time` (`time`),
            KEY `idx_mac_time` (`mac`,`time`)
            ) ENGINE=INNODB DEFAULT CHARSET=utf8 COMMENT '5分钟数据表'
    </update>

    <insert id="insertHistoryFiveMinutely">
        INSERT INTO
        history_five_minutely_${timeUnits}
        VALUES
        <foreach collection="list" item="item" separator=",">
            (#{item.mac}, #{item.time}, #{item.value})
        </foreach>
    </insert>

</mapper>

 screen-job/src/main/resources/mapper/HistoryMinutelyMapper.xml

New file
@@ -0,0 +1,50 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.HistoryMinutelyMapper">

    <update id="createTable" parameterType="String">
        CREATE TABLE  IF NOT EXISTS `history_minutely_${timeUnits}` (
            `mac` VARCHAR (20) DEFAULT NULL COMMENT '设备mac',
            `time` datetime DEFAULT NULL COMMENT '数据时间',
            `value` json DEFAULT NULL COMMENT '数据',
            `version` INT (11) DEFAULT NULL COMMENT '型号',
            KEY `idx_mac` (`mac`),
            KEY `idx_time` (`time`),
            KEY `idx_mac_time` (`mac`,`time`)
            ) ENGINE=INNODB DEFAULT CHARSET=utf8 COMMENT '分钟数据表'
    </update>

    <select id="getHistoryFiveMinutelyData" resultType="java.util.LinkedHashMap">
        SELECT
        mac
        <foreach collection="sensorKeys" open="," separator="," item="sensorKey">
            <choose>
                <when test="sensorKey=='e23[0]'">
                    ROUND((CASE WHEN AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))<![CDATA[>]]>0 AND
                    AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI()))<![CDATA[>]]>0
                    THEN
                    ATAN(AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))/AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI())))*180/PI()
                    WHEN AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))>0 AND
                    AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI()))<![CDATA[<]]>0
                    THEN
                    (ATAN(AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))/AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI())))*180/PI())+180
                    WHEN AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))<![CDATA[<]]>0 AND
                    AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI()))<![CDATA[<]]>0
                    THEN
                    (ATAN(AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))/AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI())))*180/PI())+180
                    ELSE
                    (ATAN(AVG(value->'$.e18[0]'*SIN((value->'$.e23[0]'/180)*PI()))/AVG(value->'$.e18[0]'*COS((value->'$.e23[0]'/180)*PI())))*180/PI())+360
                    END),3) AS '${sensorKey}'
                </when>
                <otherwise>
                    ROUND(AVG(value->'$.${sensorKey}[0]'),3) AS '${sensorKey}'
                </otherwise>
            </choose>
        </foreach>
        FROM
        history_minutely_${timeUnits}
        WHERE time <![CDATA[>=]]> #{start} AND time <![CDATA[<]]> #{end}
        GROUP BY mac
    </select>

</mapper>

 screen-manage/src/main/java/com/moral/api/config/kafka/KafkaConsumerConfig.java

@@ -2,7 +2,6 @@

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -13,6 +12,7 @@
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

@@ -27,8 +27,6 @@
    private String sessionTimeout;
    @Value("${kafka.consumer.auto.commit.interval}")
    private String autoCommitInterval;
    @Value("${kafka.consumer.groupMenu.id}")
    private String groupId;
    @Value("${kafka.consumer.auto.offset.reset}")
    private String autoOffsetReset;
    @Value("${kafka.consumer.concurrency}")
@@ -41,16 +39,12 @@
        factory.setConcurrency(concurrency);
        factory.getContainerProperties().setPollTimeout(1500);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        /*factory.setBatchListener(true);//@KafkaListener 批量消费  每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置提交偏移量的方式*/
        return factory;
    }

    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }


    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
@@ -60,7 +54,6 @@
        propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        return propsMap;
    }

 screen-manage/src/main/java/com/moral/api/controller/DeviceController.java

@@ -56,7 +56,7 @@
        //判断mac是否已存在
        QueryWrapper<Device> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("mac", device.getMac());
        if (deviceService.getOne(queryWrapper) != null) {
        if (deviceService.count(queryWrapper) > 0) {
            return ResultMessage.fail(ResponseCodeEnum.MAC_IS_EXIST.getCode(), ResponseCodeEnum.MAC_IS_EXIST.getMsg());
        }
        deviceService.insert(device);

 screen-manage/src/main/java/com/moral/api/controller/TestController.java

@@ -4,6 +4,7 @@
import com.moral.api.entity.Test;
import com.moral.api.service.TestService;
import com.moral.api.util.CacheUtils;
import com.moral.constant.KafkaConstants;
import com.moral.constant.ResultMessage;
import com.moral.redis.RedisUtil;
import com.moral.util.PageResult;
@@ -101,15 +102,31 @@
    }

    /**
     * kafka測試
     * 分钟主题kafka測試
     */
    @ApiOperation(value = "kafka測試", notes = "kafka測試")
    @ApiImplicitParams({
            @ApiImplicitParam(name = "data", value = "data", required = true, paramType = "query", dataType = "String"),
            @ApiImplicitParam(name = "token", value = "token", required = true, paramType = "header", dataType = "String")
    })
    @RequestMapping(value = "kafkaTest", method = RequestMethod.GET)
    public void kafkaTest() {
        kafkaTemplate.send("test_topic", "{'mac': 'p5dnd1234567','DataTime':1623058244104,'e1':10,'e2':20,'ver':2}");
    @RequestMapping(value = "minuteKafkaTest", method = RequestMethod.GET)
    public void minuteKafkaTest(String data) {
        System.out.println(data);
        kafkaTemplate.send(KafkaConstants.TOPIC_MINUTE, data);
    }

    /**
     * 小时主题kafka測試
     */
    @ApiOperation(value = "kafka測試", notes = "kafka測試")
    @ApiImplicitParams({
            @ApiImplicitParam(name = "data", value = "data", required = true, paramType = "query", dataType = "String"),
            @ApiImplicitParam(name = "token", value = "token", required = true, paramType = "header", dataType = "String")
    })
    @RequestMapping(value = "hourKafkaTest", method = RequestMethod.GET)
    public void hourKafkaTest(String data) {
        System.out.println(data);
        kafkaTemplate.send(KafkaConstants.TOPIC_HOUR, data);
    }

    @GetMapping("testToken")

 screen-manage/src/main/java/com/moral/api/entity/HistoryHourly.java

New file
@@ -0,0 +1,51 @@
package com.moral.api.entity;

import com.baomidou.mybatisplus.extension.activerecord.Model;

import java.io.Serializable;
import java.util.Date;

import lombok.Data;
import lombok.EqualsAndHashCode;

/**
 * <p>
 * 已校准小时表
 * </p>
 *
 * @author moral
 * @since 2021-06-04
 */
@Data
@EqualsAndHashCode(callSuper = false)
public class HistoryHourly extends Model<HistoryHourly> {

    private static final long serialVersionUID = 1L;

    /**
     * 设备mac
     */
    private String mac;

    /**
     * 数据时间
     */
    private Date time;

    /**
     * 数据
     */
    private String value;

    /**
     * 型号
     */
    private Integer version;


    @Override
    protected Serializable pkVal() {
        return null;
    }

}

 screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java

@@ -1,37 +1,128 @@
package com.moral.api.kafka.consumer;

import com.moral.api.constant.TopicConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Random;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;

@Component
import com.alibaba.fastjson.JSON;
import com.moral.api.service.DeviceService;
import com.moral.api.service.HistoryHourlyService;
import com.moral.api.service.HistoryMinutelyService;
import com.moral.api.util.AdjustDataUtils;
import com.moral.constant.KafkaConstants;
import com.moral.constant.RedisConstants;

//@Component
@Slf4j
public class KafkaConsumer {

/*    *//**
     * 这是手动提交的消费方式
     * @param record
     * @param ack
     * @throws Exception
     *//*
    @KafkaListener(topics = TopicConstants.TEST_TOPIC_MESSAGE,groupId = "test")
    public void listenTest(ConsumerRecord<String, String> record , Acknowledgment ack) throws Exception {
    @Autowired
    private HistoryMinutelyService historyMinutelyService;

    @Autowired
    private HistoryHourlyService historyHourlyService;

    @Autowired
    private DeviceService deviceService;

    @Autowired
    private AdjustDataUtils adjustDataUtils;

    @Autowired
    private RedisTemplate redisTemplate;

    //分钟数据
    @KafkaListener(topics = KafkaConstants.TOPIC_MINUTE, groupId = KafkaConstants.GROUP_ID_INSERT, containerFactory = "kafkaListenerContainerFactory")
    public void listenMinute(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String msg = record.value();
        System.out.println(msg);
        if (new Random().nextInt(100)<50){
            log.info(String.format("kafka 消费消息成功---------------- listen1 topic = %s, offset = %d, value = %s ", record.topic(), record.offset(), record.value()));
        try {
            Map<String, Object> data = JSON.parseObject(msg, HashMap.class);
            Object mac = data.get("mac");
            Object time = data.get("DataTime");
            Object ver = data.get("ver");
            if (StringUtils.isEmpty(ver) || StringUtils.isEmpty(time) || StringUtils.isEmpty(mac)) {
                log.warn("some properties is null, param{}", msg);
                ack.acknowledge();
                return;
            }

            //数据过滤
            data = data.entrySet().stream()
                    .filter(map -> {
                        String key = map.getKey();
                        return !(key.contains("Min") || key.contains("Max") || key.contains("Cou"));
                    }).collect(Collectors.toMap(m -> m.getKey().replaceAll("-Avg", ""), Map.Entry::getValue));
            data.remove("time");
            //存入数据库
            historyMinutelyService.insertHistoryMinutely(data);
            ack.acknowledge();
        } catch (Exception e) {
            //log.error("param{}" + msg);
        }
    }

    }*/
    //小时数据
    @KafkaListener(topics = KafkaConstants.TOPIC_HOUR, groupId = KafkaConstants.GROUP_ID_INSERT, containerFactory = "kafkaListenerContainerFactory")
    public void listenHour(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String msg = record.value();
        try {
            Map<String, Object> data = JSON.parseObject(msg, HashMap.class);
            Object mac = data.get("mac");
            Object time = data.get("DataTime");
            Object ver = data.get("ver");
            if (StringUtils.isEmpty(ver) || StringUtils.isEmpty(time) || StringUtils.isEmpty(mac)) {
                log.warn("some properties is null, param{}", msg);
                ack.acknowledge();
                return;
            }

            //数据过滤
            data = data.entrySet().stream()
                    .filter(map -> {
                        String key = map.getKey();
                        return !(key.contains("Min") || key.contains("Max") || key.contains("Cou"));
                    }).collect(Collectors.toMap(m -> m.getKey().replaceAll("-Avg", ""), Map.Entry::getValue));
            data.remove("time");
            //存入数据库
            historyHourlyService.insertHistoryHourly(data);
            ack.acknowledge();
        } catch (Exception e) {
            //log.error("param{}" + msg);
        }
    }




    //秒数据，修改设备状态，缓存最新秒数据
    @KafkaListener(topics = KafkaConstants.TOPIC_SECOND, groupId = KafkaConstants.GROUP_ID_STATE, containerFactory = "kafkaListenerContainerFactory")
    public void listenSecond(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String msg = record.value();
        try {
            Map<String, Object> data = JSON.parseObject(msg, HashMap.class);
            Object mac = data.get("mac");
            Object time = data.get("DataTime");
            Object ver = data.get("ver");
            if (StringUtils.isEmpty(ver) || StringUtils.isEmpty(time) || StringUtils.isEmpty(mac)) {
                log.warn("some properties is null, param{}", msg);
                ack.acknowledge();
                return;
            }
            //数据校准
            data = adjustDataUtils.adjust(data);
            //存入redis
            redisTemplate.opsForValue().set(RedisConstants.DEVICE_DATA + "_" + mac, data);
            //判断并修改设备状态
            deviceService.judgeDeviceState(data);
            ack.acknowledge();
        } catch (Exception e) {
            //log.error("param{}" + msg);
        }
    }
}

 screen-manage/src/main/java/com/moral/api/mapper/HistoryHourlyMapper.java

New file
@@ -0,0 +1,20 @@
package com.moral.api.mapper;

import java.util.Map;

import com.moral.api.entity.HistoryHourly;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;

/**
 * <p>
 * 小时表 Mapper 接口
 * </p>
 *
 * @author moral
 * @since 2021-06-04
 */
public interface HistoryHourlyMapper extends BaseMapper<HistoryHourly> {

    void insertHistoryHourlyUnAdjust(Map<String, Object> params);

}

 screen-manage/src/main/java/com/moral/api/service/DeviceService.java

@@ -56,9 +56,9 @@
    Map<String, Object> getDeviceByMac(String mac);

    //设备数据校准
    Map<String, Object> adjustDeviceData(Map<String, Object> deviceData, Map<String, Object> deviceInfo);
    Map<String, Object> adjustDeviceData(Map<String, Object> deviceData);

    //判断并缓存设备状态
    Map<String, Object> judgeDeviceState(Map<String, Object> data, Map<String, Object> deviceInfo);
    //判断并修改设备状态
    void judgeDeviceState(Map<String, Object> data);

}

 screen-manage/src/main/java/com/moral/api/service/HistoryHourlyService.java

New file
@@ -0,0 +1,20 @@
package com.moral.api.service;

import java.util.Map;

import com.moral.api.entity.HistoryHourly;
import com.baomidou.mybatisplus.extension.service.IService;

/**
 * <p>
 * 小时表 服务类
 * </p>
 *
 * @author moral
 * @since 2021-06-04
 */
public interface HistoryHourlyService extends IService<HistoryHourly> {

    void insertHistoryHourly(Map<String, Object> data);

}

 screen-manage/src/main/java/com/moral/api/service/HistoryMinutelyService.java

@@ -13,6 +13,6 @@
 */
public interface HistoryMinutelyService {

    void insertHistoryMinutely(Map<String, Object> deviceData);
    void insertHistoryMinutely(Map<String, Object> data);

}

 screen-manage/src/main/java/com/moral/api/service/impl/DeviceServiceImpl.java

@@ -1,26 +1,31 @@
package com.moral.api.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.googlecode.aviator.AviatorEvaluator;
import com.googlecode.aviator.Expression;
import com.moral.api.entity.*;
import com.moral.api.mapper.*;
import com.moral.api.pojo.vo.device.DeviceVO;
import com.moral.api.service.DeviceService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;

import com.moral.api.util.AdjustDataUtils;
import com.moral.api.util.LogUtils;
import com.moral.constant.Constants;
import com.moral.constant.RedisConstants;
import com.moral.util.ConvertUtils;
import com.moral.util.DateUtils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

@@ -28,10 +33,12 @@
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import javax.servlet.http.HttpServletRequest;

@@ -44,6 +51,7 @@
 * @since 2021-05-11
 */
@Service
@Slf4j
public class DeviceServiceImpl extends ServiceImpl<DeviceMapper, Device> implements DeviceService {

    @Autowired
@@ -76,13 +84,14 @@
    @Autowired
    private VersionSensorUnitMapper versionSensorUnitMapper;


    @Autowired
    private AdjustDataUtils adjustDataUtils;

    /*
     * 从redis获取设备信息
     * */
    private Map<String, Object> getDeviceInfoFromRedis(String mac) {
        Map<String, Object> deviceInfo = (Map<String, Object>) redisTemplate.opsForValue().get(RedisConstants.DEVICE + mac);
        Map<String, Object> deviceInfo = (Map<String, Object>) redisTemplate.opsForValue().get(getDeviceKey(mac));
        return deviceInfo;
    }

@@ -107,6 +116,13 @@
        return keysConnect(RedisConstants.DEVICE, mac);
    }

    /*
     * 从redis获取报警级别
     * */
    private Map<String, Object> getOrgAlarmConfigFromRedis(String orId) {
        return null;
    }

    //redis key前缀
    private String keysConnect(String... keys) {
        StringBuilder key = new StringBuilder(keys[0]);
@@ -125,7 +141,7 @@
        deviceMapper.insert(device);
        Map<String, Object> deviceInfo = selectDeviceInfoById(device.getId());
        //维护组织型号关系表
        insertOrganizationUnitAlarm(orgId,device.getDeviceVersionId());
        insertOrganizationUnitAlarm(orgId, device.getDeviceVersionId());
        //新增设备信息存入redis
        String mac = device.getMac();
        //从redis中删除设备信息
@@ -142,17 +158,13 @@
    @Override
    @Transactional
    public void delete(Integer deviceId) {
        Device device = deviceMapper.selectById(deviceId);
        UpdateWrapper<Device> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id", deviceId).set("is_delete", Constants.DELETE);
        deviceMapper.update(null, updateWrapper);
        Device device = deviceMapper.selectById(deviceId);
        String mac = device.getMac();
        //从redis中删除设备信息
        delDeviceInfoFromRedis(mac);
        //维护组织型号关系表
        Integer versionId = device.getDeviceVersionId();
        Integer orgId = device.getOrganizationId();
        deleteOrganizationUnitAlarm(orgId,versionId);
        //操作日志记录
        HttpServletRequest request = ((ServletRequestAttributes) Objects.requireNonNull(RequestContextHolder.getRequestAttributes())).getRequest();
        StringBuilder content = new StringBuilder();
@@ -166,7 +178,7 @@
        Integer deviceId = device.getId();
        Device oldDevice = deviceMapper.selectById(deviceId);
        //判断是否更新了站点，如果更新了站点则查询对应站点的组织id进行更新
        if(!ObjectUtils.isEmpty(device.getMonitorPointId())){
        if (!ObjectUtils.isEmpty(device.getMonitorPointId())) {
            MonitorPoint monitorPoint = monitorPointMapper.selectById(device.getMonitorPointId());
            device.setOrganizationId(monitorPoint.getOrganizationId());
        }
@@ -174,14 +186,14 @@
        Device updateDevice = deviceMapper.selectById(deviceId);
        String mac = updateDevice.getMac();
        //维护组织型号关系表
       Integer oldOrgId = oldDevice.getOrganizationId();
       Integer newOrgId = updateDevice.getOrganizationId();
       Integer oldVersionId = oldDevice.getDeviceVersionId();
       Integer newVersionId = updateDevice.getDeviceVersionId();
       if(!oldOrgId.equals(newOrgId)||!oldVersionId.equals(newVersionId)){
           deleteOrganizationUnitAlarm(oldOrgId,oldVersionId);
           insertOrganizationUnitAlarm(newOrgId,newVersionId);
       }
        Integer oldOrgId = oldDevice.getOrganizationId();
        Integer newOrgId = updateDevice.getOrganizationId();
        Integer oldVersionId = oldDevice.getDeviceVersionId();
        Integer newVersionId = updateDevice.getDeviceVersionId();
        if (!oldOrgId.equals(newOrgId) || !oldVersionId.equals(newVersionId)) {
            deleteOrganizationUnitAlarm(oldOrgId, oldVersionId);
            insertOrganizationUnitAlarm(newOrgId, newVersionId);
        }
        //从redis中删除设备信息
        delDeviceInfoFromRedis(mac);
        Map<String, Object> deviceInfo = selectDeviceInfoById(deviceId);
@@ -276,13 +288,15 @@
            queryWrapper.eq("monitor_point_id", mpId);
        }

        //设备名模糊查询
        if (name != null) {
            queryWrapper.like("name", name);
        }

        //mac模糊查询
        if (mac != null) {
            queryWrapper.like("mac", mac);
        }


        //排序参数,默认create_time降序
        if (order != null && orderType != null) {
@@ -311,7 +325,6 @@
        result.put("item", items);
        return result;
    }


    @Override
    public Map<String, Object> selectDeviceInfoById(Integer deviceId) {
@@ -382,6 +395,9 @@
        MonitorPoint monitorPoint = device.getMonitorPoint();
        mpInfo.put("id", monitorPoint.getId());
        mpInfo.put("name", monitorPoint.getName());
        mpInfo.put("areaCode", monitorPoint.getAreaCode());
        mpInfo.put("cityCode", monitorPoint.getCityCode());
        mpInfo.put("provinceCode", monitorPoint.getProvinceCode());
        deviceInfo.put("monitorPoint", mpInfo);

        setDeviceInfoToRedis(mac, deviceInfo);
@@ -418,27 +434,74 @@
    }

    @Override
    public Map<String, Object> adjustDeviceData(Map<String, Object> deviceData, Map<String, Object> deviceInfo) {
        return null;
    public Map<String, Object> adjustDeviceData(Map<String, Object> deviceData) {
        return adjustDataUtils.adjust(deviceData);
    }

    @Override
    public Map<String, Object> judgeDeviceState(Map<String, Object> deviceData, Map<String, Object> deviceInfo) {
        return null;
    public void judgeDeviceState(Map<String, Object> deviceData) {
        String mac = deviceData.remove("mac").toString();
        Device device = (Device) redisTemplate.opsForHash().get(RedisConstants.DEVICE_INFO, mac);
        Version version = device.getVersion();
        List<Sensor> sensors = version.getSensors();
        Expression expression;
        int state = 1;
        for (Sensor sensor : sensors) {
            //因子报警等级
            String alarmLevel = sensor.getAlarmLevel();
            if (StringUtils.isEmpty(alarmLevel)) {
                continue;
            }
            List<Double> list = JSONObject.parseObject(alarmLevel, List.class);
            String sensorCode = sensor.getCode();
            //转换公式
            String formula = sensor.getFormula();
            //转换单位前因子值
            String sensorValue = (String) deviceData.get(sensorCode);
            double value = Double.parseDouble(String.format("%.3f", sensorValue));
            //单位转换
            if (formula != null) {
                //转换后因子值
                sensorValue = formula.replace("{0}", sensorValue);
                expression = AviatorEvaluator.compile(sensorValue);
                value = Double.parseDouble(String.format("%.3f", expression.execute()));
            }
            int sensorState = judgeState(list, value);
            if (sensorState > state) {
                state = sensorState;
            }
        }
        //修改设备状态
        UpdateWrapper<Device> updateWrapper = new UpdateWrapper<>();
        updateWrapper.eq("id", device.getId()).set("state", state);
        deviceMapper.update(null, updateWrapper);
    }

    private void insertOrganizationUnitAlarm(Integer orgId,Integer versionId){
    //根据因子值判断状态
    private int judgeState(List<Double> levels, Double data) {
        int state = 1;
        for (int i = levels.size() - 1; i >= 0; i--) {
            Double level = levels.get(i);
            if (data >= level) {
                state = i + 2;
                break;
            }
        }
        return state;
    }

    private void insertOrganizationUnitAlarm(Integer orgId, Integer versionId) {
        QueryWrapper<OrganizationUnitAlarm> queryOrganizationVersionWrapper = new QueryWrapper<>();
        queryOrganizationVersionWrapper.eq("organization_id",orgId);
        queryOrganizationVersionWrapper.eq("version_id",versionId);
        queryOrganizationVersionWrapper.eq("is_delete",Constants.NOT_DELETE);
        queryOrganizationVersionWrapper.eq("organization_id", orgId);
        queryOrganizationVersionWrapper.eq("version_id", versionId);
        queryOrganizationVersionWrapper.eq("is_delete", Constants.NOT_DELETE);
        List<OrganizationUnitAlarm> organizationUnitAlarms = organizationUnitAlarmMapper.selectList(queryOrganizationVersionWrapper);
        if(ObjectUtils.isEmpty(organizationUnitAlarms)){
            QueryWrapper<VersionSensorUnit> queryVersionSensorUnitWrapper =new QueryWrapper<>();
            queryVersionSensorUnitWrapper.eq("version_id",versionId);
            queryVersionSensorUnitWrapper.eq("is_delete",Constants.NOT_DELETE);
        if (ObjectUtils.isEmpty(organizationUnitAlarms)) {
            QueryWrapper<VersionSensorUnit> queryVersionSensorUnitWrapper = new QueryWrapper<>();
            queryVersionSensorUnitWrapper.eq("version_id", versionId);
            queryVersionSensorUnitWrapper.eq("is_delete", Constants.NOT_DELETE);
            List<VersionSensorUnit> versionSensorUnits = versionSensorUnitMapper.selectList(queryVersionSensorUnitWrapper);
            if(!ObjectUtils.isEmpty(versionSensorUnits)){
            if (!ObjectUtils.isEmpty(versionSensorUnits)) {
                for (VersionSensorUnit versionSensorUnit : versionSensorUnits) {
                    OrganizationUnitAlarm organizationUnitAlarm = new OrganizationUnitAlarm();
                    organizationUnitAlarm.setOrganizationId(orgId);
@@ -452,19 +515,19 @@
        }
    }

    private void deleteOrganizationUnitAlarm(Integer orgId,Integer versionId){
    private void deleteOrganizationUnitAlarm(Integer orgId, Integer versionId) {
        QueryWrapper<Device> queryOrganizationVersionWrapper = new QueryWrapper<>();
        queryOrganizationVersionWrapper.eq("organization_id",orgId);
        queryOrganizationVersionWrapper.eq("device_version_id",versionId);
        queryOrganizationVersionWrapper.eq("is_delete",Constants.NOT_DELETE);
        queryOrganizationVersionWrapper.eq("organization_id", orgId);
        queryOrganizationVersionWrapper.eq("device_version_id", versionId);
        queryOrganizationVersionWrapper.eq("is_delete", Constants.NOT_DELETE);
        List<Device> devices = deviceMapper.selectList(queryOrganizationVersionWrapper);
        if(ObjectUtils.isEmpty(devices)){//如果为空，则组织没有该型号的设备了。
        if (ObjectUtils.isEmpty(devices)) {//如果为空，则组织没有该型号的设备了。
            UpdateWrapper deleteWrapper = new UpdateWrapper();
            deleteWrapper.eq("organization_id",orgId);
            deleteWrapper.eq("version_id",versionId);
            deleteWrapper.eq("is_delete",Constants.NOT_DELETE);
            deleteWrapper.set("is_delete",Constants.DELETE);
            organizationUnitAlarmMapper.update(null,deleteWrapper);
            deleteWrapper.eq("organization_id", orgId);
            deleteWrapper.eq("version_id", versionId);
            deleteWrapper.eq("is_delete", Constants.NOT_DELETE);
            deleteWrapper.set("is_delete", Constants.DELETE);
            organizationUnitAlarmMapper.update(null, deleteWrapper);
        }
    }
}

 screen-manage/src/main/java/com/moral/api/service/impl/HistoryHourlyServiceImpl.java

New file
@@ -0,0 +1,63 @@
package com.moral.api.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.moral.api.entity.HistoryHourly;
import com.moral.api.mapper.HistoryHourlyMapper;
import com.moral.api.service.DeviceService;
import com.moral.api.service.HistoryHourlyService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.moral.constant.Constants;
import com.moral.util.DateUtils;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * <p>
 * 小时表 服务实现类
 * </p>
 *
 * @author moral
 * @since 2021-06-04
 */
@Service
public class HistoryHourlyServiceImpl extends ServiceImpl<HistoryHourlyMapper, HistoryHourly> implements HistoryHourlyService {

    @Autowired
    private HistoryHourlyMapper historyHourlyMapper;

    @Autowired
    private DeviceService deviceService;

    @Override
    public void insertHistoryHourly(Map<String, Object> data) {
        Map<String, Object> dataAdjust = new HashMap<>(data);
        String mac = data.remove("mac").toString();
        Date time = DateUtils.dataToTimeStampTime(new Date(new Long((String) data.remove("DataTime"))), DateUtils.yyyy_MM_dd_HH_EN);
        Integer version = (Integer) data.remove("ver");
        Map<String, Object> result = new HashMap<>(data);
        result.put("mac", mac);
        result.put("time", time);
        result.put("version", version);
        result.put("timeUnits", Constants.UN_ADJUST);
        result.put("value", JSONObject.toJSONString(data));
        //原始数据（未校准）
        historyHourlyMapper.insertHistoryHourlyUnAdjust(result);

        //数据校准
        dataAdjust = deviceService.adjustDeviceData(dataAdjust);
        dataAdjust.remove("mac");
        dataAdjust.remove("DataTime");
        dataAdjust.remove("ver");
        HistoryHourly historyHourly = new HistoryHourly();
        historyHourly.setMac(mac);
        historyHourly.setTime(time);
        historyHourly.setVersion(version);
        historyHourly.setValue(JSONObject.toJSONString(dataAdjust));
        historyHourlyMapper.insert(historyHourly);
    }
}

 screen-manage/src/main/java/com/moral/api/service/impl/HistoryMinutelyServiceImpl.java

@@ -7,7 +7,7 @@
import java.util.HashMap;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.moral.api.mapper.HistoryMinutelyMapper;
import com.moral.api.service.DeviceService;
import com.moral.api.service.HistoryMinutelyService;
@@ -24,30 +24,31 @@
    private DeviceService deviceService;

    @Override
    public void insertHistoryMinutely(Map<String, Object> deviceData) {
    public void insertHistoryMinutely(Map<String, Object> data) {
        Map<String, Object> result = new HashMap<>();
        Object mac = deviceData.remove("mac");
        Map<String, Object> dataAdjust = new HashMap<>(data);
        Object mac = data.remove("mac");
        result.put("mac", mac);
        result.put("version", deviceData.remove("ver"));
        Date time = new Date((Long) deviceData.remove("DataTime"));
        result.put("time", DateUtils.dateToDateString(time));
        result.put("value", JSON.toJSONString(deviceData));
        result.put("version", data.remove("ver"));
        Date time = new Date(new Long((String) data.remove("DataTime")));
        result.put("time", DateUtils.dataToTimeStampTime(time,DateUtils.yyyy_MM_dd_HH_mm_EN));
        result.put("value", JSONObject.toJSONString(data));
        String timeUnits = DateUtils.dateToDateString(time, DateUtils.yyyyMM_EN);
        result.put("timeUnits", tableSuffix(timeUnits, Constants.UN_ADJUST));
        //未校准

        //原始数据（未校准）
        historyMinutelyMapper.insertHistoryMinutely(result);

        //设备信息
        Map<String, Object> deviceInfo = deviceService.getDeviceByMac(mac.toString());

        //设备数据校准，并存入数据库
        //数据校准
        dataAdjust = deviceService.adjustDeviceData(dataAdjust);
        dataAdjust.remove("mac");
        dataAdjust.remove("DataTime");
        dataAdjust.remove("ver");
        result.put("timeUnits", timeUnits);
        deviceData = deviceService.adjustDeviceData(deviceData, deviceInfo);
        result.put("value", JSON.toJSONString(deviceData));
        result.put("value", JSONObject.toJSONString(dataAdjust));
        historyMinutelyMapper.insertHistoryMinutely(result);

        //判断设备状态
        Map<String, Object> deviceState = deviceService.judgeDeviceState(deviceData, deviceInfo);

    }

    //表后缀
@@ -59,5 +60,4 @@
        }
        return key.toString();
    }

}

 screen-manage/src/main/java/com/moral/api/util/AdjustDataUtils.java

New file
@@ -0,0 +1,97 @@
package com.moral.api.util;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.googlecode.aviator.AviatorEvaluator;
import com.googlecode.aviator.Expression;
import com.moral.api.entity.DeviceAdjustValue;
import com.moral.api.service.DeviceService;
import com.moral.constant.RedisConstants;
import com.moral.util.DateUtils;

@Slf4j
@Component
public class AdjustDataUtils {

    @Autowired
    private DeviceService deviceService;

    @Autowired
    private RedisTemplate redisTemplate;

    public Map<String, Object> adjust(Map<String, Object> deviceData) {
        try {
            Object dataTime = deviceData.get("DataTime");
            String mac = deviceData.get("mac").toString();
            //清除毫秒，四舍五入
            long time = Math.round(new Double((String) dataTime) / 1000) * 1000L;
            long finalTime = DateUtils.dataToTimeStampTime(new Date(time), DateUtils.HH_mm_ss_EN).getTime();
            //设备信息
            Map<String, Object> deviceInfo = deviceService.getDeviceByMac(mac);
            Map<String, Object> monitorPoint = (Map<String, Object>) deviceInfo.get("monitorPoint");
            Object areaCode = monitorPoint.get("areaCode");
            Object cityCode = monitorPoint.get("cityCode");
            for (String key : deviceData.keySet()) {
                if (!key.equals("mac") && !key.equals("time") && !key.equals("DataTime") && !key.equals("ver") && !key.contains("Flag")) {
                    //测量值
                    Object measuredValue = deviceData.get(key);
                    List<DeviceAdjustValue> adjustValues = (List<DeviceAdjustValue>) redisTemplate.opsForHash().get(RedisConstants.ADJUST + "_" + mac, key);
                    if (ObjectUtils.isEmpty(adjustValues)) {
                        deviceData.put(key, measuredValue);
                        continue;
                    }

                    //根据时间段筛选校准公式
                    DeviceAdjustValue deviceAdjustValue = adjustValues.stream()
                            .filter(o -> o.getStartTime().getTime() <= finalTime && o.getEndTime().getTime() > finalTime)
                            .findFirst().get();
                    String adjustValue = deviceAdjustValue.getValue();
                    if (ObjectUtils.isEmpty(adjustValue)) {
                        deviceData.put(key, measuredValue);
                        continue;
                    }

                    Expression expression = AviatorEvaluator.compile(adjustValue);
                    Map<String, Object> env = new HashMap<>();
                    if (adjustValue.contains("aqi")) {
                        Object aqiValue = redisTemplate.opsForHash().get("aqi_" + areaCode, key);
                        if (ObjectUtils.isEmpty(aqiValue)) {
                            aqiValue = redisTemplate.opsForHash().get("aqi_" + cityCode, key);
                        }
                        env.put("aqi", ObjectUtils.isEmpty(aqiValue) ? 0F : Float.parseFloat((String) aqiValue));
                    }
                    if (adjustValue.contains("vocs")) {
                        Object vocsValue = ObjectUtils.isEmpty(deviceData.get("a99054")) ? 0F : deviceData.get("a99054");
                        env.put("vocs", vocsValue);
                    }
                    if (adjustValue.contains("cel")) {
                        env.put("cel", Float.parseFloat((String) measuredValue));
                    }
                    //校准
                    measuredValue = expression.execute(env);
                    //温度处理
                    if (Float.parseFloat(measuredValue.toString()) < 0 && !"a01001".equals(measuredValue)) {
                        measuredValue = 0F;
                    }
                    deviceData.put(key, Double.parseDouble(String.format("%.3f", measuredValue)));

                }
            }
        } catch (Exception e) {
            log.error("param[0] deviceData:" + JSON.toJSONString(deviceData));
            log.error(e.getMessage());
        }
        return deviceData;
    }

}

 screen-manage/src/main/resources/application-dev.yml

@@ -92,12 +92,9 @@
    enable:
      auto:
        commit: false
    group:
      id: test
    servers: 192.168.0.16:9092,192.168.0.17:9092,192.168.0.18:9092
    session:
      timeout: 6000
    topic: test_topic
    zookeeper:
      connect: 192.168.0.16:2181,192.168.0.17:2181,192.168.0.18:2181
  producer:

 screen-manage/src/main/resources/mapper/DeviceMapper.xml

@@ -49,7 +49,9 @@
        <association property="monitorPoint" javaType="com.moral.api.entity.MonitorPoint">
            <result column="mp_id" property="id"/>
            <result column="mp_name" property="name"/>
            <result column="mp_address" property="address"/>
            <result column="mp_area_code" property="areaCode"/>
            <result column="mp_city_code" property="cityCode"/>
            <result column="mp_province_code" property="provinceCode"/>
        </association>

        <!--型号-->
@@ -116,7 +118,9 @@
            o.`name` org_name,
            mp.id mp_id,
            mp.`name` mp_name,
            mp.address mp_address,
            mp.area_code mp_area_code,
            mp.city_code mp_city_code,
            mp.province_code mp_province_code,
            ma.id operate_id,
            v.id version_id,
            v.`name` version_name,

 screen-manage/src/main/resources/mapper/HistoryHourlyMapper.xml

New file
@@ -0,0 +1,17 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.moral.api.mapper.HistoryHourlyMapper">

    <!-- 通用查询映射结果 -->
    <resultMap id="BaseResultMap" type="com.moral.api.entity.HistoryHourly">
        <result column="mac" property="mac"/>
        <result column="time" property="time"/>
        <result column="value" property="value"/>
        <result column="version" property="version"/>
    </resultMap>

    <insert id="insertHistoryHourlyUnAdjust">
        INSERT INTO history_hourly_${timeUnits} VALUES (#{mac}, #{time}, #{value}, #{version})
    </insert>

</mapper>