Merge remote-tracking branch 'origin/master'

# Conflicts:
#	smart-city-platform/src/main/resources/mapper/DeviceDataDimMapper.xml

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/IotDataProcessorApplication.java

@@ -4,7 +4,9 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cache.annotation.EnableCaching;
 import org.springframework.context.annotation.ComponentScan;
+import org.springframework.scheduling.annotation.EnableAsync;
 
 /**
  * IOT数据处理器
@@ -12,6 +14,8 @@ import org.springframework.context.annotation.ComponentScan;
  */
 @Slf4j
 @SpringBootApplication
+@EnableAsync
+@EnableCaching
 @ComponentScan(basePackages={"com.zcxk"})
 public class IotDataProcessorApplication implements CommandLineRunner {
 

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/config/AsyncConfiguration.java

@@ -0,0 +1,35 @@
+package com.zcxk.zoniot.data.processor.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import java.util.concurrent.Executor;
+
+/**
+ * <p>异步任务配置</p>
+ *
+ * @Author wilian.peng
+ * @Date 2020/7/6 15:46
+ * @Version 1.0
+ */
+@Configuration
+public class AsyncConfiguration {
+
+    @Bean("taskExecutor")
+    public Executor asyncExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        //核心线程数8：线程池创建时候初始化的线程数
+        executor.setCorePoolSize(8);
+        //最大线程数16：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程
+        executor.setMaxPoolSize(16);
+        //缓冲队列500：用来缓冲执行任务的队列
+        executor.setQueueCapacity(500);
+        //允许线程的空闲时间60秒：当超过了核心线程出之外的线程在空闲时间到达之后会被销毁
+        executor.setKeepAliveSeconds(200);
+        //线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池
+        executor.setThreadNamePrefix("ProcessAsync-");
+        executor.initialize();
+        return executor;
+    }
+}

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/listener/DeviceDataMessageBus.java

@@ -1,10 +1,12 @@
 package com.zcxk.zoniot.data.processor.listener;
 
+import com.zcxk.zoniot.data.processor.service.AsyncProcessDeviceDataService;
 import com.zcxk.zoniot.data.processor.service.DeviceDataProcessor;
 import com.zcxk.zoniot.smartcity.common.dto.DeviceDataDTO;
 import com.zcxk.zoniot.smartcity.common.model.DeviceData;
 import com.zcxk.zoniot.smartcity.common.utils.DtoEntityUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.StopWatch;
 import org.bson.Document;
 import org.bson.types.ObjectId;
 import org.springframework.beans.factory.InitializingBean;
@@ -14,6 +16,7 @@ import org.springframework.data.mongodb.core.query.BasicQuery;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
 import java.util.Collections;
@@ -34,15 +37,23 @@ public class DeviceDataMessageBus implements InitializingBean {
 
     @Autowired
     MongoTemplate mongoTemplate ;
+
+    @Autowired
+    AsyncProcessDeviceDataService asyncProcessDeviceDataService;
+
     @KafkaListener(topics = {"${device.data.topic}"})
     public void receiveDeviceDataMessage(String message) {
+        log.info("Receive DeviceDataMessage Method , Message = {}",message);
         String[] splits = message.split("#");
         Query query = new Query();
         query.addCriteria(Criteria.where("_id").is(new ObjectId(splits[0])));
         query.addCriteria(Criteria.where("deviceId").is(Long.parseLong(splits[1])));
         DeviceData data = mongoTemplate.findOne(query, DeviceData.class);
-        for(DeviceDataProcessor processor : deviceDataProcessors){
-            processor.process(data);
+        asyncProcessDeviceDataService.asyncProcessDeviceData(data,deviceDataProcessors);
+        try {
+            Thread.sleep(100L);
+        } catch (InterruptedException e) {
+            log.error("thread interrupt",e);
         }
     }
 

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/model/WaterMeterType.java

@@ -1,5 +1,6 @@
 package com.zcxk.zoniot.data.processor.model;
 
+import java.io.Serializable;
 import java.util.Date;
 import lombok.Data;
 
@@ -10,7 +11,7 @@ import lombok.Data;
  * @Version 1.0
  */
 @Data
-public class WaterMeterType {
+public class WaterMeterType implements Serializable {
     /**
     * 主键
     */

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/service/AsyncProcessDeviceDataService.java

@@ -0,0 +1,32 @@
+package com.zcxk.zoniot.data.processor.service;
+
+import com.zcxk.zoniot.smartcity.common.model.DeviceData;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.StopWatch;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * <p></p>
+ *
+ * @Author wilian.peng
+ * @Date 2020/8/7 21:28
+ * @Version 1.0
+ */
+@Slf4j
+@Component
+public class AsyncProcessDeviceDataService {
+    @Async
+    public void asyncProcessDeviceData(DeviceData data ,List<DeviceDataProcessor> deviceDataProcessors){
+        log.info("Async Process DeviceDataMessage , DeviceId = {},Data={}",data.getDeviceId(),data.getId());
+        StopWatch clock = new StopWatch();
+        clock.start(); // 计时开始
+        for(DeviceDataProcessor processor : deviceDataProcessors){
+            processor.process(data);
+        }
+        clock.stop(); // 计时结束
+        log.info("Async Process DeviceDataMessage , DeviceId = {} ,Cost:{}ms",data.getDeviceId(),clock.getTime());
+    }
+}

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/service/impl/AlarmProcessor.java

@@ -7,6 +7,7 @@ import com.zcxk.zoniot.data.processor.service.DeviceErrorService;
 import com.zcxk.zoniot.smartcity.common.dto.DeviceDataDTO;
 import com.zcxk.zoniot.smartcity.common.model.DeviceData;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.StopWatch;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -30,12 +31,15 @@ public class AlarmProcessor extends DeviceDataProcessor {
     }
     @Override
     public void process(DeviceData data) {
-        log.debug("Begin Execute AlarmProcessor , Data = {}",data.getId());
+        log.info("Begin Execute AlarmProcessor , DeviceId = {},Data = {}",data.getDeviceId(),data.getId());
+        StopWatch clock = new StopWatch();
+        clock.start(); // 计时开始
         try {
             deviceErrorService.processDeviceData(data);
         }catch (Exception e){
             log.error("Execute AlarmProcessor Failed !",e);
         }
-        log.debug("End Execute AlarmProcessor , Data = {}",data.getId());
+        clock.stop(); // 计时结束
+        log.info("End Execute AlarmProcessor , DeviceId = {},Cost:{}ms",data.getDeviceId(),clock.getTime());
     }
 }

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/service/impl/CacheDataProcessor.java

@@ -9,6 +9,7 @@ import com.zcxk.zoniot.data.processor.utils.RedisUtil;
 import com.zcxk.zoniot.smartcity.common.model.DeviceData;
 import com.zcxk.zoniot.smartcity.common.model.DeviceMeasuringData;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.StopWatch;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -44,7 +45,9 @@ public class CacheDataProcessor extends DeviceDataProcessor {
     }
     @Override
     public void process(DeviceData data) {
-        log.debug("Begin Execute CacheDataProcessor , Data = {}",data.getId());
+        log.info("Begin Execute CacheDataProcessor , DeviceId = {},Data = {}",data.getDeviceId(),data.getId());
+        StopWatch clock = new StopWatch();
+        clock.start(); // 计时开始
         try{
             // 1,添加最新上报时间的测点
             Date receiveTime = data.getReceiveTime();
@@ -66,7 +69,8 @@ public class CacheDataProcessor extends DeviceDataProcessor {
         }catch (Exception e){
             log.error("Execute CacheDataProcessor Failed !",e);
         }
-        log.debug("End Execute CacheDataProcessor , Data = {}",data.getId());
+        clock.stop(); // 计时结束
+        log.info("End Execute CacheDataProcessor , DeviceId = {},Cost:{}ms",data.getDeviceId(),clock.getTime());
     }
 
     protected String formatDate(Date date) {

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/service/impl/DeviceServiceImpl.java

@@ -30,7 +30,7 @@ public class DeviceServiceImpl implements DeviceService {
     @Autowired
     DeviceMapper deviceMapper;
     
-    @Cacheable(value = "WaterMeterDeviceCache",key = "#deviceId",unless = "#result == null")
+    //@Cacheable(value = "WaterMeterDeviceCache",key = "#deviceId",unless = "#result == null")
     @Override
     public MeterDeviceDTO findWaterMeter(Long deviceId,Integer deviceTypeId) {
         MeterDeviceDTO meterDevice = null ;

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/service/impl/RealTimeDataPushProcessor.java

@@ -5,6 +5,7 @@ import com.zcxk.zoniot.data.processor.service.DeviceDataPushService;
 import com.zcxk.zoniot.smartcity.common.dto.DeviceDataDTO;
 import com.zcxk.zoniot.smartcity.common.model.DeviceData;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.time.StopWatch;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -26,12 +27,15 @@ public class RealTimeDataPushProcessor extends DeviceDataProcessor {
     }
     @Override
     public void process(DeviceData data) {
-        log.debug("Begin Execute RealTimeDataPushProcessor , Data = {}",data.getId());
+        log.info("Begin Execute RealTimeDataPushProcessor , DeviceId = {},Data = {}",data.getDeviceId(),data.getId());
+        StopWatch clock = new StopWatch();
+        clock.start(); // 计时开始
         try {
             deviceDataPushService.realTimeDataPush(data);
         }catch (Exception e){
             log.error("Excute RealTimeDataPushProcessor Failed !",e);
         }
-        log.debug("End Execute RealTimeDataPushProcessor , Data = {}",data.getId());
+        clock.stop(); // 计时结束
+        log.info("End Execute RealTimeDataPushProcessor , DeviceId = {},Cost:{}ms",data.getDeviceId(),clock.getTime());
     }
 }

iot-data-processor/src/main/java/com/zcxk/zoniot/data/processor/service/impl/WaterMeterDataProcessor.java

@@ -10,6 +10,7 @@ import com.zcxk.zoniot.smartcity.common.model.DeviceMeasuringData;
 import com.zcxk.zoniot.smartcity.common.model.MeterReadRecord;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.time.StopWatch;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mongodb.core.MongoTemplate;
@@ -56,7 +57,9 @@ public class WaterMeterDataProcessor extends DeviceDataProcessor{
     }
     @Override
     public void process(DeviceData data) {
-        log.debug("Begin Execute WaterMeterDataProcessor , Data = {}",data.getId());
+        log.info("Begin Execute WaterMeterDataProcessor , DeviceId = {},Data = {}",data.getDeviceId(),data.getId());
+        StopWatch clock = new StopWatch();
+        clock.start(); // 计时开始
         try{
             // 1,判断当前设备是否为水表
             Long deviceId = data.getDeviceId();
@@ -88,7 +91,8 @@ public class WaterMeterDataProcessor extends DeviceDataProcessor{
         }catch (Exception e){
             log.error("Execute WaterMeterDataProcessor Failed !",e);
         }
-        log.debug("End Execute WaterMeterDataProcessor , Data = {}",data.getId());
+        clock.stop(); // 计时结束
+        log.info("End Execute WaterMeterDataProcessor , DeviceId = {},Cost:{}ms",data.getDeviceId(),clock.getTime());
     }
 
     /**

iot-data-processor/src/main/resources/application-sit.properties

@@ -30,7 +30,7 @@ mybatis.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl
 mybatis.configuration.use-column-label=true
 ###########################################MongoDB配置##############################################
 spring.data.mongodb.uri=mongodb://114.135.61.188:17017/water-iot-database
-logging.level.org.springframework.data.mongodb.core=DEBUG
+logging.level.org.springframework.data.mongodb.core=INFO
 #############################################Redis配置##############################################
 spring.redis.host=114.135.61.188
 spring.redis.port=26379

smart-city-dispatcher/src/main/resources/application-sit.properties

@@ -1,4 +1,5 @@
-server.port=8091
+# deploy host 10.0.0.156
+server.port=8000
 #log
 logging.level.root=info
 logging.path=/opt/sit/dispatcher/logs
@@ -6,10 +7,10 @@ logging.path=/opt/sit/dispatcher/logs
 
 
 #kafka 
-spring.kafka.bootstrap-servers=193.112.139.7:9092
+spring.kafka.bootstrap-servers=10.0.0.161:6377
 #kafka producer
 spring.kafka.producer.retries=0
 spring.kafka.producer.batch-size=4096
 spring.kafka.producer.buffer-memory=40960
-com.zcxk.kafka.single.data.topic=sc_single_data_topic
-com.zcxk.kafka.multi.data.topic=sc_multi_data_topic
+com.zcxk.kafka.single.data.topic=sc_single_data_topic_sit
+com.zcxk.kafka.multi.data.topic=sc_multi_data_topic_sit

smart-city-intf/src/main/resources/application.properties

@@ -1,2 +1,2 @@
 #开发环境：dev  测试环境：sit  线上环境：prd
-spring.profiles.active=uat
+spring.profiles.active=sit