screen-api kafka消费者组id从配置文件获取 screen-manage ...

kaiyu

2021-07-28 11d10fc9ca90ec29159738c1eac5e2d9203d8bb6

screen-api
                kafka消费者组id从配置文件获取
screen-manage
                除admin角色外无法查看admin角色

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

@@ -29,6 +29,8 @@
    private String autoOffsetReset;
    @Value("${kafka.consumer.concurrency}")
    private int concurrency;
    @Value("${kafka.groupId.second-data}")
    private String secondDataGroupId;

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
@@ -42,8 +44,51 @@
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }
    
    @Bean("secondDataListenerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> secondDataListenerFactory(){
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(secondDataConsumerFactory());//设置消费者工厂
        factory.setConcurrency(concurrency);//设置线程数
        factory.getContainerProperties().setPollTimeout(1500);//设置拉取数据超时时间
        return factory;
    }

    /**
    * @Description: 秒级消费工厂
            * @Param: []
            * @return: org.springframework.kafka.core.ConsumerFactory<java.lang.String,java.lang.String>
            * @Author: 陈凯裕
            * @Date: 2021/7/19
            */
    public ConsumerFactory<String,String> secondDataConsumerFactory(){
        Map<String, Object> commonConfig = consumerConfigs();
        Map<String, Object> secondDataConfig = secondConsumerConfigs();
        secondDataConfig.putAll(commonConfig);
        return new DefaultKafkaConsumerFactory<>(secondDataConfig);
    }

    /**
    * @Description: 秒级消费者配置
            * @Param: []
            * @return: java.util.Map<java.lang.String,java.lang.Object>
            * @Author: 陈凯裕
            * @Date: 2021/7/19
            */
    public Map<String,Object> secondConsumerConfigs(){
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.GROUP_ID_CONFIG,secondDataGroupId);
        return propsMap;
    }


    /**
    * @Description: 通用配置
            * @Param: []
            * @return: java.util.Map<java.lang.String,java.lang.Object>
            * @Author: 陈凯裕
            * @Date: 2021/7/19
            */
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);

 screen-api/src/main/java/com/moral/api/controller/HistoryFiveMinutelyController.java

@@ -36,11 +36,11 @@

    @GetMapping("queryDeviceAndData")
    public ResultMessage queryDeviceAndData(QueryDeviceAndFiveMinuteDataForm form){
        //18 110100
        //判断是否缺少参数
        if (!form.valid())
            return ResultMessage.fail(ResponseCodeEnum.PARAMETERS_IS_MISSING.getCode(),
                    ResponseCodeEnum.PARAMETERS_IS_MISSING.getMsg());

        //处理查询业务
        List<DeviceAndFiveMinuteDataDTO> dto = historyFiveMinutelyService.queryDeviceAndFiveMinuteData(form);


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

@@ -6,9 +6,12 @@
import com.moral.api.entity.UnitConversion;
import com.moral.api.websocket.SingleDeviceServer;
import com.moral.util.UnitConvertUtils;
import lombok.AllArgsConstructor;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.stereotype.Component;
@@ -29,7 +32,7 @@
@Component
public class SecondDataConsumer implements ConsumerSeekAware {

    @KafkaListener(topics = "second_data", groupId = "SecondsDataGroup996")
    @KafkaListener(containerFactory = "secondDataListenerFactory",topics = "second_data")
    public void listen(ConsumerRecord<String, String> record, Consumer consumer) throws Exception {
        String messageStr = record.value();
        Map<String, Object> message = (Map<String, Object>) JSON.parse(messageStr);
@@ -39,7 +42,7 @@
            //判断消息书否数据该socket
            String messageMac = (String) message.get("mac");
            String mac = socket.getMac();
            if(!mac.equalsIgnoreCase(messageMac))
            if (!mac.equalsIgnoreCase(messageMac))
                continue;
            //取出基本信息
            Map<String, Device> devicesInfo = socket.getDevicesInfo();
@@ -70,22 +73,22 @@
                    //单位转换
                    String resultData = UnitConvertUtils.calculate(sourceData, formula);
                    if (resultData != null) {
                        resultData+=showUnit;
                    }else{//如果转换出的数据为null，则代表缓存中也没有公式，依然使用源单位。
                        resultData = sourceData+unit;
                        resultData += showUnit;
                    } else {//如果转换出的数据为null，则代表缓存中也没有公式，依然使用源单位。
                        resultData = sourceData + unit;
                    }
                    resultMessgae.put(sensor.getCode(), resultData);
                } else {
                    //拼接单位
                    sourceData += showUnit;
                    sourceData = sourceData + " " + showUnit;
                    resultMessgae.put(sensor.getCode(), sourceData);
                }
            }
            resultMessgae.put("DataTime",message.get("DataTime"));
            resultMessgae.put("time",message.get("time"));
            resultMessgae.put("time1",message.get("time1"));
            resultMessgae.put("serverTime",System.currentTimeMillis());
            resultMessgae.put("serverStartTime",serverStartTime);
            resultMessgae.put("DataTime", message.get("DataTime"));
            resultMessgae.put("time", message.get("time"));
            resultMessgae.put("time1", message.get("time1"));
            resultMessgae.put("serverTime", System.currentTimeMillis());
            resultMessgae.put("serverStartTime", serverStartTime);
            socket.sendMessage(JSON.toJSONString(resultMessgae));
        }
    }

 screen-api/src/main/java/com/moral/api/pojo/form/historyFiveMinutely/QueryDeviceAndFiveMinuteDataForm.java

@@ -26,4 +26,6 @@
            return false;
        return true;
    }


}

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

@@ -27,6 +27,15 @@
            */
    List<DeviceAndFiveMinuteDataDTO> queryDeviceAndFiveMinuteData(QueryDeviceAndFiveMinuteDataForm form);

    /**
    * @Description: 查询一个设备最新的五分钟数据
            * @Param: [mac]
            * @return: java.util.List<com.moral.api.entity.HistoryFiveMinutely>
            * @Author: 陈凯裕
            * @Date: 2021/7/23
            */
    HistoryFiveMinutely queryLastDataByMac(String mac);

    //获取5分钟风场数据
    List<Object> getAreaWindData(Map<String,Object> params);
}

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

@@ -40,8 +40,8 @@
    @Override
    public List<Device> getDevicesByMonitorPointId(Integer monitorPointId) {
        QueryWrapper<Device> wrapper = new QueryWrapper();
        wrapper.eq("monitor_point_id",monitorPointId);
        wrapper.eq("is_delete", Constants.NOT_DELETE);
        wrapper.eq("monitor_point_id", monitorPointId);
        List<Device> devices = deviceMapper.selectList(wrapper);
        return devices;
    }

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

@@ -1,6 +1,8 @@
package com.moral.api.service.impl;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.moral.api.entity.Device;
import com.moral.api.entity.HistoryFiveMinutely;
import com.moral.api.entity.MonitorPoint;
@@ -14,19 +16,13 @@
import com.moral.api.service.MonitorPointService;
import com.moral.constant.RedisConstants;
import com.moral.util.DateUtils;

import io.lettuce.core.GeoCoordinates;
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 java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.*;

/**
 * <p>
@@ -41,12 +37,10 @@

    @Autowired
    MonitorPointService monitorPointService;

    @Autowired
    RedisTemplate redisTemplate;

    @Autowired
    private HistoryFiveMinutelyMapper historyFiveMinutelyMapper;
    HistoryFiveMinutelyMapper historyFiveMinutelyMapper;

    @Override
    public List<DeviceAndFiveMinuteDataDTO> queryDeviceAndFiveMinuteData(QueryDeviceAndFiveMinuteDataForm form) {
@@ -67,18 +61,36 @@
        for (Device device : devices) {
            DeviceAndFiveMinuteDataDTO dto = new DeviceAndFiveMinuteDataDTO();
            String mac = device.getMac();
            //从缓存中获取
            Map<String, Object> sensorValues = (Map<String, Object>) redisTemplate.opsForHash().get(RedisConstants.DATA_FIVE_MINUTES, mac);
            Map<String, Object> value = new HashMap<>();
            if (sensorValues != null && sensorValues.get(sensorCode) != null)
                value.put(sensorCode, sensorValues.get(sensorCode));
            //如果没有数据从数据库查询
            HistoryFiveMinutely dbHistoryFiveMinutely = queryLastDataByMac(mac);
            String dbDataStr = dbHistoryFiveMinutely.getValue();
            sensorValues = JSON.parseObject(dbDataStr, HashMap.class);
            Map<String,Object> sensorValue = new HashMap<>();
            if (sensorValues!=null&&sensorValues.get(sensorCode) != null)
                sensorValue.put(sensorCode,sensorValues.get(sensorCode));
            else
                value.put(sensorCode, null);
                sensorValue.put(sensorCode,null);
            dto.setDevice(device);
            dto.setSensorValue(value);
            dto.setSensorValue(sensorValue);
            dtos.add(dto);
        }
        return dtos;
    }

    @Override
    public HistoryFiveMinutely queryLastDataByMac(String mac) {
        QueryWrapper<HistoryFiveMinutely> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("mac",mac);
        queryWrapper.orderByDesc("time");
        queryWrapper.last("limit 0,1");
        List<HistoryFiveMinutely> historyFiveMinutelies = historyFiveMinutelyMapper.selectList(queryWrapper);
        if(ObjectUtils.isEmpty(historyFiveMinutelies))
            return null;
        return historyFiveMinutelies.get(0);
    }


    @Override
    public List<Object> getAreaWindData(Map<String, Object> params) {
@@ -272,4 +284,8 @@
        list.add(laLaMap);
        return list;
    }




}

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

@@ -105,6 +105,8 @@
    linger: 1
    retries: 0
    servers: 172.16.44.65:9092,172.16.44.67:9092,172.16.44.66:9092
  groupId:
    second-data: SecondsDataGroup
mvc:
  interceptor:
    exclude:

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

@@ -40,6 +40,7 @@
    public void judgeOffLineDevice() {
        QueryWrapper<Device> queryWrapper = new QueryWrapper<>();
        queryWrapper.ne("state", Constants.DEVICE_STATE_OFFLINE);
        queryWrapper.eq("is_delete",Constants.NOT_DELETE);
        //获取所有在线设备
        List<Device> devices = deviceMapper.selectList(queryWrapper);
        for (Device device : devices) {
@@ -64,4 +65,4 @@
        updateWrapper.eq("id", device.getId()).set("state", device.getState());
        deviceMapper.update(null, updateWrapper);
    }
}
}

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

@@ -4,11 +4,13 @@
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.moral.api.entity.*;
import com.moral.api.mapper.*;
import com.moral.api.pojo.redisBean.AccountInfoDTO;
import com.moral.api.service.ManageRoleService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.moral.api.util.LogUtils;
import com.moral.constant.Constants;
import com.moral.constant.ResponseCodeEnum;
import com.moral.util.TokenUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -21,6 +23,7 @@
import javax.servlet.http.HttpServletRequest;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.function.Predicate;

/**
 * <p>
@@ -79,6 +82,19 @@
            manageRoleMap.put("key",manageRole.getId());
            manageRolesList.add(manageRoleMap);
        }
        //除admin账号外  admin角色不可见
        AccountInfoDTO accountInfo = (AccountInfoDTO) TokenUtils.getUserInfo();
        List<ManageRole> roles = accountInfo.getRoles();
        if(!roles.get(0).getName().equals("admin")){
            manageRolesList.removeIf(new Predicate<Map<String, Object>>() {
                @Override
                public boolean test(Map<String, Object> map) {
                    if(map.get("name").equals("admin"))
                        return true;
                    return false;
                }
            });
        }
        resultMap.put("manageRoles",manageRolesList);
        int totalNumber = manageRoleMapper.selectCount(wrapper);
        resultMap.put("totalNumber",totalNumber);