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() { @@ -43,7 +45,50 @@ 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);
@@ -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); @@ -77,7 +80,7 @@ resultMessgae.put(sensor.getCode(), resultData); } else { //拼接单位 sourceData += showUnit; sourceData = sourceData + " " + showUnit; resultMessgae.put(sensor.getCode(), sourceData); } } 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("is_delete", Constants.NOT_DELETE); wrapper.eq("monitor_point_id", monitorPointId); wrapper.eq("is_delete", Constants.NOT_DELETE); 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<>(); //如果没有数据从数据库查询 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) value.put(sensorCode, sensorValues.get(sensorCode)); 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) { 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);
