screen-api/src/main/java/com/moral/api/entity/UnitConversion.java
New file @@ -0,0 +1,74 @@ 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 java.util.Date; import lombok.Data; import lombok.EqualsAndHashCode; /** * <p> * 单位转换表 * </p> * * @author moral * @since 2021-07-12 */ @Data @EqualsAndHashCode(callSuper = false) public class UnitConversion extends Model<UnitConversion> { private static final long serialVersionUID = 1L; /** * 主键 */ @TableId(value = "id", type = IdType.AUTO) private Integer id; /** * 原始单位键 */ private String originalUnitKey; /** * 目标单位键 */ private Integer targetUnitKey; /** * 转换公式 */ private String formula; /** * 当为null时所有因子都使用。如果有因子code，则该公式只适用于这个因子 */ private String sensorCode; /** * 创建时间 */ private Date createTime; /** * 更新时间 */ private Date updateTime; /** * 是否删除 */ private String isDelete; @Override protected Serializable pkVal() { return this.id; } } screen-api/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java
@@ -1,13 +1,7 @@ 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.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Component; import java.util.Random; @Component @Slf4j @@ -27,11 +21,5 @@ log.info(String.format("kafka 消费消息成功---------------- listen1 topic = %s, offset = %d, value = %s ", record.topic(), record.offset(), record.value())); ack.acknowledge(); } }*/ } screen-api/src/main/java/com/moral/api/kafka/consumer/SecondDataConsumer.java
@@ -3,6 +3,7 @@ import com.alibaba.fastjson.JSON; import com.moral.api.entity.Device; import com.moral.api.entity.Sensor; import com.moral.api.entity.UnitConversion; import com.moral.api.websocket.SingleDeviceServer; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; @@ -10,6 +11,7 @@ import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.listener.ConsumerSeekAware; import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; import java.util.HashMap; import java.util.List; @@ -30,7 +32,6 @@ public void listen(ConsumerRecord<String, String> record , Consumer consumer) throws Exception { String messageStr = record.value(); Map<String,String> message = (Map<String,String>)JSON.parse(messageStr); //System.out.println(message); CopyOnWriteArraySet<SingleDeviceServer> sockets = SingleDeviceServer.sockets; for (SingleDeviceServer socket : sockets) { //判断消息书否数据该socket @@ -44,9 +45,21 @@ List<Sensor> sensors = device.getVersion().getSensors();//获取型号所有因子信息 for (Sensor sensor : sensors) { String code = sensor.getCode(); String showUnit = sensor.getShowUnit(); String showUnitKey = sensor.getShowUnitKey(); String unitKey = sensor.getUnitKey(); //数据补偿 //单位转换 if(!unitKey.equals(showUnitKey)){ String formula = sensor.getFormula(); if(ObjectUtils.isEmpty(formula)){ List<UnitConversion> unitConversions = socket.getUnitConversions(); System.out.println(unitConversions); }else{ } } //拼接单位 String showUnit = sensor.getShowUnit(); String value = message.get(code); value+=showUnit; resultMessgae.put(sensor.getCode(),value); @@ -72,3 +85,4 @@ } }
New file @@ -0,0 +1,16 @@ package com.moral.api.mapper; import com.moral.api.entity.UnitConversion; import com.baomidou.mybatisplus.core.mapper.BaseMapper; /** * <p> * 单位转换表 Mapper 接口 * </p> * * @author moral * @since 2021-07-12 */ public interface UnitConversionMapper extends BaseMapper<UnitConversion> { } screen-api/src/main/java/com/moral/api/service/UnitConversionService.java
New file @@ -0,0 +1,16 @@ package com.moral.api.service; import com.moral.api.entity.UnitConversion; import com.baomidou.mybatisplus.extension.service.IService; /** * <p> * 单位转换表 服务类 * </p> * * @author moral * @since 2021-07-12 */ public interface UnitConversionService extends IService<UnitConversion> { } screen-api/src/main/java/com/moral/api/service/impl/UnitConversionServiceImpl.java
New file @@ -0,0 +1,20 @@ package com.moral.api.service.impl; import com.moral.api.entity.UnitConversion; import com.moral.api.mapper.UnitConversionMapper; import com.moral.api.service.UnitConversionService; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import org.springframework.stereotype.Service; /** * <p> * 单位转换表 服务实现类 * </p> * * @author moral * @since 2021-07-12 */ @Service public class UnitConversionServiceImpl extends ServiceImpl<UnitConversionMapper, UnitConversion> implements UnitConversionService { } screen-api/src/main/java/com/moral/api/websocket/SingleDeviceServer.java
@@ -2,6 +2,7 @@ import com.moral.api.entity.Device; import com.moral.api.entity.Sensor; import com.moral.api.entity.UnitConversion; import com.moral.constant.RedisConstants; import lombok.Data; import org.springframework.data.redis.core.RedisTemplate; @@ -30,22 +31,25 @@ //线程安全集合，用于存放server对象 public static CopyOnWriteArraySet<SingleDeviceServer> sockets = new CopyOnWriteArraySet<>(); public static RedisTemplate redisTemplate; private Session session; private String mac; public static RedisTemplate redisTemplate; private Map<String, Device> devicesInfo; private Map<String,Device> devicesInfo; private List<UnitConversion> unitConversions; @OnOpen public void onOpen(Session session, @PathParam("mac") String mac) throws Exception { this.session = session; this.mac = mac; this.devicesInfo = redisTemplate.opsForHash().entries(RedisConstants.DEVICE_INFO); this.unitConversions = redisTemplate.opsForList().range(RedisConstants.UNIT_CONVERSION, 0, -1); sockets.add(this); System.out.println(mac); } } @OnClose public void onClose() { @@ -63,11 +67,10 @@ public void sendMessage(String message) throws Exception { if (this.session.isOpen()) { synchronized (session) { this.session.getBasicRemote().sendText(message); } // synchronized (session) { this.session.getBasicRemote().sendText(message); // } } } } screen-api/src/main/resources/application-local.yml
New file @@ -0,0 +1,119 @@ server: port: 8081 tomcat: uri-encoding: UTF-8 #最小线程数 min-spare-threads: 500 #最大线程数 max-threads: 2500 #最大链接数 max-connections: 6500 #最大等待队列长度 accept-count: 1000 spring: profiles: active: dev application: name: screen-manage redis: host: r-bp1xdlb9wfc6zt0msppd.redis.rds.aliyuncs.com port: 6379 password: moral_123456 timeout: 30000 jedis: pool: max-active: 256 max-wait: 30000 max-idle: 64 min-idle: 32 lettuce: pool: max-active: 256 max-idle: 64 max-wait: 30000 min-idle: 32 tokenRedis: host: r-bp1xdlb9wfc6zt0msppd.redis.rds.aliyuncs.com port: 6379 password: moral_123456 timeout: 30000 database: 15 pool: max-active: 256 max-wait: 30000 max-idle: 64 min-idle: 32 datasource: minIdle: 1 time-between-eviction-runs-millis: 60000 max-active: 20 test-while-idle: true validation-query: select 'x' filters: stat type: com.alibaba.druid.pool.DruidDataSource max-wait: 60000 url: jdbc:mysql://rm-bp1pr3rx9m3fnkwsk8o.mysql.rds.aliyuncs.com:3306/moral?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=Asia/Shanghai username: root password: moral_123456 test-on-borrow: false sql-script-encoding: utf-8 pool-prepared-statements: true min-evictable-idle-time-millis: 300000 initial-size: 1 driver-class-name: com.mysql.cj.jdbc.Driver max-conn-lifetime-millis: 20 test-on-return: false mybatis-plus: mapper-locations: classpath:mapper/*.xml global-config: db-config: id-type: auto field-strategy: NOT_EMPTY db-type: MYSQL configuration: map-underscore-to-camel-case: true call-setters-on-nulls: true log-impl: org.apache.ibatis.logging.stdout.StdOutImpl kafka: listener: ack-mode: manual_immediate consumer: auto: offset: reset: latest concurrency: 3 enable: auto: commit: false servers: 47.99.145.48:9092,47.111.116.38:9092,47.111.124.159:9092 session: timeout: 6000 zookeeper: connect: 47.99.145.48:2181,47.111.116.38:2181,47.111.124.159:2181 producer: batch: size: 4096 buffer: memory: 40960 linger: 1 retries: 0 servers: 47.99.145.48:9092,47.111.116.38:9092,47.111.124.159:9092 mvc: interceptor: exclude: - /login - /swagger-ui.html - /swagger-resources/** - /swagger-ui.html/** - /webjars/** - /verificationCode/** AES: KEY: AD42F7787B035B7580000EF93BE20BAD TOKEN: KEY: foh3wi2ooghiCh5 screen-api/src/main/resources/mapper/UnitConversionMapper.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.UnitConversionMapper"> <!-- 通用查询映射结果 --> <resultMap id="BaseResultMap" type="com.moral.api.entity.UnitConversion"> <id column="id" property="id" /> <result column="original_unit_key" property="originalUnitKey" /> <result column="target_unit_key" property="targetUnitKey" /> <result column="formula" property="formula" /> <result column="sensor_code" property="sensorCode" /> <result column="create_time" property="createTime" /> <result column="update_time" property="updateTime" /> <result column="is_delete" property="isDelete" /> </resultMap> </mapper> screen-common/src/main/java/com/moral/constant/RedisConstants.java
@@ -58,5 +58,11 @@ * */ public static final String AQI_DATA = "aqi_data"; /* * 单位转换公式 * 接收类型List<UnitConversion> * */ public static final String UNIT_CONVERSION = "unit_conversion"; } screen-manage/src/main/java/com/moral/api/controller/TestController.java
@@ -4,12 +4,14 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.moral.api.entity.Sensor; import com.moral.api.entity.Test; import com.moral.api.entity.UnitConversion; import com.moral.api.service.SensorService; import com.moral.api.service.TestService; import com.moral.api.service.impl.SensorServiceImpl; import com.moral.api.util.CacheUtils; import com.moral.constant.Constants; import com.moral.constant.KafkaConstants; import com.moral.constant.RedisConstants; import com.moral.constant.ResultMessage; import com.moral.redis.RedisUtil; import com.moral.util.PageResult; @@ -22,6 +24,7 @@ import io.swagger.annotations.ApiOperation; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.web.bind.annotation.*; @@ -42,6 +45,9 @@ @Resource private TestService testService; @Autowired RedisTemplate redisTemplate; /** * name 姓名 @@ -206,4 +212,10 @@ } } @GetMapping("test") public void test(){ List<UnitConversion> range = redisTemplate.opsForList().range(RedisConstants.UNIT_CONVERSION, 0, -1); System.out.println(range); } } screen-manage/src/main/java/com/moral/api/kafka/consumer/KafkaConsumer.java
@@ -20,7 +20,7 @@ import com.moral.constant.KafkaConstants; import com.moral.constant.RedisConstants; @Component //@Component @Slf4j public class KafkaConsumer {
@@ -55,4 +55,13 @@ */ UnitConversionDTO deleteUnitConversion(UnitConversionDeleteForm form); /** * @Description: 刷新单位缓存 * @Param: [] * @return: void * @Author: 陈凯裕 * @Date: 2021/7/12 */ void refreshCache(); } screen-manage/src/main/java/com/moral/api/service/impl/UnitConversionServiceImpl.java
@@ -18,9 +18,11 @@ import com.moral.api.service.UnitConversionService; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.moral.constant.Constants; import com.moral.constant.RedisConstants; import com.moral.constant.ResponseCodeEnum; import com.moral.util.ConvertUtils; 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; @@ -44,6 +46,8 @@ UnitConversionMapper unitConversionMapper; @Autowired SysDictDataService sysDictDataService; @Autowired RedisTemplate redisTemplate; @Override @Transactional @@ -76,6 +80,8 @@ } //插入数据 unitConversionMapper.insert(unitConversion); //刷新缓存 refreshCache(); //封装返回数据 dto.setCode(ResponseCodeEnum.SUCCESS.getCode()); dto.setMsg(ResponseCodeEnum.SUCCESS.getMsg()); @@ -153,6 +159,8 @@ oldUnitConversion = unitConversionMapper.selectOne(oldWrapper); //更新 unitConversionMapper.updateById(unitConversion); //刷新缓存 refreshCache(); //封装返回结果 dto.setCode(ResponseCodeEnum.SUCCESS.getCode()); dto.setMsg(ResponseCodeEnum.SUCCESS.getMsg()); @@ -177,15 +185,21 @@ deleteWrapper.set("is_delete", Constants.DELETE); deleteWrapper.eq("id", id); unitConversionMapper.update(null, deleteWrapper); //刷新缓存 refreshCache(); //封装返回对象 dto.setCode(ResponseCodeEnum.SUCCESS.getCode()); dto.setMsg(ResponseCodeEnum.SUCCESS.getMsg()); return dto; } public static void main(String[] args) { int[] array = new int[]{1,2,3,4,5,}; String s = JSON.toJSONString(array); System.out.println(s); /*刷新缓存*/ @Override public void refreshCache(){ redisTemplate.delete((RedisConstants.UNIT_CONVERSION)); QueryWrapper<UnitConversion> queryWrapper = new QueryWrapper<>(); queryWrapper.eq("is_delete",Constants.NOT_DELETE); List<UnitConversion> unitConversions = unitConversionMapper.selectList(queryWrapper); redisTemplate.opsForList().leftPushAll(RedisConstants.UNIT_CONVERSION,unitConversions); } } screen-manage/src/main/java/com/moral/api/util/CacheUtils.java
@@ -7,6 +7,7 @@ import com.moral.api.mapper.UnitConversionMapper; import com.moral.api.service.SensorService; import com.moral.api.service.SysDictDataService; import com.moral.api.service.UnitConversionService; import com.moral.constant.Constants; import com.moral.constant.RedisConstants; import com.moral.util.TokenUtils; @@ -40,6 +41,11 @@ private static UnitConversionMapper unitConversionMapper; private static SensorService sensorService; private static UnitConversionService unitConversionService; @Autowired public void setUnitConversionService(UnitConversionService unitConversionService){CacheUtils.unitConversionService = unitConversionService;} @Autowired public void setSensorService(SensorService sensorService) { @@ -123,4 +129,8 @@ public static void refreshDicTypeAndData(){ } public static void refreshUnitConversion(){ unitConversionService.refreshCache(); } }
