xxl

zoniot-rmcp/zoniot-rmcp-core/src/main/java/com/zcxk/rmcp/core/mapper/DeviceMapper.xml

@@ -764,7 +764,7 @@
     left join rmcp_org rod on (rod.ID = rd.dept_org_id)
     left join rmcp_community rc on (rc.id = rd.community_id)
     where rd.status = 1
-    order by rd.id asc
+    order by rd.id asc limit #{startIndex},#{pageSize}
   </select>
 
   <select id="queryDeviceInfoList" resultType="com.zcxk.rmcp.api.vo.DeviceVo">

zoniot-rmcp/zoniot-xxljob-client/src/main/java/com/zcxk/xxljob/common/Constants.java

@@ -15,6 +15,11 @@ public class Constants {
      */
     public static final String PREFIX_CACHE_FLAG = "last-meter-data:";
 
+    /**
+     * 设备最后数据缓存前缀
+     */
+    public static final String PREFIX_DEVICE_CACHE_FLAG = "last-device-data:";
+
     /**
      * 抄表记录表名称
      */

zoniot-rmcp/zoniot-xxljob-client/src/main/java/com/zcxk/xxljob/config/ResourceServerConfig.java

@@ -0,0 +1,31 @@
+package com.zcxk.xxljob.config;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.oauth2.config.annotation.web.configuration.EnableResourceServer;
+import org.springframework.security.oauth2.config.annotation.web.configuration.ResourceServerConfigurerAdapter;
+
+/**
+ * @author Andy
+ * @version V1.0
+ **/
+@Configuration
+@EnableResourceServer
+public class ResourceServerConfig extends ResourceServerConfigurerAdapter {
+
+    @Override
+    public void configure(HttpSecurity http) throws Exception {
+        http
+            .csrf().disable()
+            .requestMatchers().antMatchers("/**")
+            .and()
+            .authorizeRequests()
+            .antMatchers("/swagger-ui.html","/webjars/**", "/swagger-resources/**",
+                    "/test/**",
+                   "/v2/**")
+            .permitAll() //配置不需要身份认证的请求路径
+           // .anyRequest().authenticated() //其他所有访问路径都需要身份认证
+            .and()
+            .httpBasic();
+    }
+}

zoniot-rmcp/zoniot-xxljob-client/src/main/java/com/zcxk/xxljob/config/SwaggerConfig.java

@@ -0,0 +1,74 @@
+package com.zcxk.xxljob.config;
+
+import org.springframework.boot.autoconfigure.web.servlet.error.BasicErrorController;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import springfox.documentation.builders.ApiInfoBuilder;
+import springfox.documentation.builders.PathSelectors;
+import springfox.documentation.builders.RequestHandlerSelectors;
+import springfox.documentation.service.ApiInfo;
+import springfox.documentation.service.ApiKey;
+import springfox.documentation.service.AuthorizationScope;
+import springfox.documentation.service.SecurityReference;
+import springfox.documentation.spi.DocumentationType;
+import springfox.documentation.spi.service.contexts.SecurityContext;
+import springfox.documentation.spring.web.plugins.Docket;
+import springfox.documentation.swagger2.annotations.EnableSwagger2;
+
+import java.util.List;
+
+import static com.google.common.collect.Lists.newArrayList;
+
+/**
+ * @author Andy
+ * @version V1.0
+ **/
+@Configuration
+@EnableSwagger2
+public class SwaggerConfig {
+    @Bean
+    public Docket api() {
+        return new Docket(DocumentationType.SWAGGER_2)
+                .ignoredParameterTypes(BasicErrorController.class)
+                .groupName("api")
+                .select()
+                .apis(RequestHandlerSelectors.any())
+                .paths(PathSelectors.any())
+                .build()
+                .apiInfo(apiInfo())
+                .securitySchemes(securitySchemes())
+                .securityContexts(securityContexts());
+    }
+
+
+    private ApiInfo apiInfo() {
+        return new ApiInfoBuilder()
+                .title("计量集抄Api")
+                .description("计量集抄")
+                .version("1.0")
+                .build();
+    }
+
+    private List<ApiKey> securitySchemes() {
+        return newArrayList(
+                new ApiKey("Authorization", "Authorization", "header"));
+    }
+
+    private List<SecurityContext> securityContexts() {
+        return newArrayList(
+                SecurityContext.builder()
+                        .securityReferences(defaultAuth())
+                        .forPaths(PathSelectors.regex("^(?!auth).*$"))
+                        .build()
+        );
+    }
+
+    List<SecurityReference> defaultAuth() {
+        AuthorizationScope authorizationScope = new AuthorizationScope("global", "accessEverything");
+        AuthorizationScope[] authorizationScopes = new AuthorizationScope[1];
+        authorizationScopes[0] = authorizationScope;
+        return newArrayList(
+                new SecurityReference("Authorization", authorizationScopes));
+    }
+
+}

zoniot-rmcp/zoniot-xxljob-client/src/main/java/com/zcxk/xxljob/controller/TestController.java

@@ -0,0 +1,27 @@
+package com.zcxk.xxljob.controller;
+
+import com.zcxk.xxljob.jobs.MeterReadJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author linqingwei
+ * @date 2021-08-20 11:11
+ */
+@Slf4j
+@RequestMapping("test")
+@RestController
+public class TestController {
+
+    @Autowired
+    private MeterReadJob meterReadJob;
+
+    @PostMapping("readMeterTest")
+    public String readMeterTest(){
+        meterReadJob.meterReadJobHandler(null);
+        return "ok";
+    }
+}

zoniot-rmcp/zoniot-xxljob-client/src/main/java/com/zcxk/xxljob/jobs/MeterReadJob.java

@@ -45,6 +45,9 @@ public class MeterReadJob {
 
         // 分片参数
         ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
+        if (shardingVO == null) {
+            shardingVO = new ShardingUtil.ShardingVO(0,1);
+        }
         XxlJobLogger.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());
 
         // 业务逻辑

zoniot-rmcp/zoniot-xxljob-client/src/main/java/com/zcxk/xxljob/service/impl/MeterReadRecordServiceImpl.java

@@ -1,9 +1,11 @@
 package com.zcxk.xxljob.service.impl;
 
+import cn.hutool.core.comparator.CompareUtil;
 import cn.hutool.core.date.LocalDateTimeUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.PageUtil;
 import com.alibaba.fastjson.JSON;
+import com.xxl.job.core.log.XxlJobLogger;
 import com.zcxk.core.common.enums.StatusEnum;
 import com.zcxk.core.common.util.SnowflakeIdWorker;
 import com.zcxk.core.utils.DateUtil;
@@ -22,15 +24,13 @@ import org.springframework.data.mongodb.core.aggregation.*;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
 import java.math.BigDecimal;
 import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 /**
  * @author linqingwei
@@ -48,6 +48,8 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
     private SnowflakeIdWorker idWorker;
     @Resource
     private MeterReadRecordMapper meterReadRecordMapper;
+    @Autowired
+    private RedisTemplate redisTemplate;
 
     @Override
     public void executeCreateMeterUnReadRecord(Map<String, Object> map, int index, int total) {
@@ -59,34 +61,92 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
         int startIndex = index * pageSize;
 
         List<WaterMeterDto> waterMeterList = deviceMapper.findWaterMeterListWithPage(startIndex, pageSize);
+
+        XxlJobLogger.log("waterMeterList size = {}",waterMeterList.size());
         batchCreateMeterUnReadRecord(waterMeterList,readDay);
         log.info("end executeCreateMeterUnReadRecord");
     }
 
     private void batchCreateMeterUnReadRecord(List<WaterMeterDto> waterMeterList, int readDay) {
         if(waterMeterList.isEmpty())return;
+
+        updateReadingData(waterMeterList,readDay);
+
+        saveReadingData(readDay);
+    }
+
+    /**
+     * @description 更新抄表数据
+     * @param waterMeterList
+     * @return
+     * @author linqingwei
+     **/
+    private void updateReadingData(List<WaterMeterDto> waterMeterList,int readDay) {
+
+
+
+        Map<String,WaterMeterDto> lastMap = redisTemplate.opsForHash().entries(Constants.PREFIX_DEVICE_CACHE_FLAG);
+
+        List<MeterReadRecord> meterReadRecordList = new ArrayList<>();
+
+        for (WaterMeterDto waterMeter : waterMeterList) {
+            WaterMeterDto last = lastMap.get(waterMeter.getId().toString());
+            if (last != null) {
+                //判断最后和最新信息不同则更新
+                if(CompareUtil.compare(waterMeter, last) == -1){
+                    log.info("begin updateReadingData compare waterMeter = {} last = {} {}",JSON.toJSONString(waterMeter),JSON.toJSONString(last),readDay);
+                    Query query = new Query();
+                    query.addCriteria(Criteria.where("deviceId").is(waterMeter.getId()));
+                    Update update = new Update();
+                    if(waterMeter.getTenantId()!=null)update.set("tenantId",waterMeter.getTenantId());
+                    if(waterMeter.getCategoryId()!=null)update.set("categoryId",waterMeter.getCategoryId());
+                    if(waterMeter.getCompanyOrgId()!=null)update.set("companyOrgId",waterMeter.getCompanyOrgId());
+                    if(waterMeter.getCompanyName()!=null)update.set("companyOrgName",waterMeter.getCompanyName());
+                    if(waterMeter.getDeptOrgId()!=null)update.set("deptOrgId",waterMeter.getDeptOrgId());
+                    if(waterMeter.getDeptName()!=null)update.set("deptOrgName",waterMeter.getDeptName());
+                    if(waterMeter.getCommunityId()!=null)update.set("communityId",waterMeter.getCommunityId());
+                    if(waterMeter.getCommunityName()!=null)update.set("communityName",waterMeter.getCommunityName());
+                    if(waterMeter.getDeviceNo()!=null)update.set("deviceNo",waterMeter.getDeviceNo());
+                    if(waterMeter.getAddress()!=null)update.set("location",waterMeter.getAddress());
+                    if(waterMeter.getProductName()!=null)update.set("deviceModel",waterMeter.getManufacturerName()+"/"+waterMeter.getProductName()+"/"+waterMeter.getProductModel());
+                    if(waterMeter.getMeterNo()!=null)update.set("meterNo",waterMeter.getMeterNo());
+                    if(waterMeter.getFileNo()!=null)update.set("fileNo",waterMeter.getFileNo());
+                    long updateResult = mongoTemplate.updateMulti(query,update, Constants.METER_READ_RECORD_TABLE).getModifiedCount();
+                }
+
+            }else {
+                meterReadRecordList.add(buildMeterReadRecord(waterMeter));
+            }
+
+
+        }
+        XxlJobLogger.log("meterReadRecordList size = {},lastMap size = {}",meterReadRecordList.size() , lastMap.size());
+        if(!meterReadRecordList.isEmpty()){
+            mongoTemplate.insert(meterReadRecordList,Constants.METER_READ_RECORD_TABLE);
+        }
+
+        //保存最新信息到缓存中
+        Map<String,WaterMeterDto> map = new HashMap<>();
         for (WaterMeterDto waterMeter : waterMeterList) {
-            saveReadingData(waterMeter,readDay);
+            map.put(waterMeter.getId().toString(),waterMeter);
         }
+        redisTemplate.opsForHash().putAll(Constants.PREFIX_DEVICE_CACHE_FLAG,map);
+
+
     }
 
-    private void saveReadingData(WaterMeterDto waterMeter,int readDay) {
-        log.info("begin saveReadingData ,waterMeter = {}" , JSON.toJSONString(waterMeter));
+    private void saveReadingData(int readDay) {
+        log.info("begin saveReadingData ,readDay = {}" , readDay);
         MeterReadRecord.MeterReadInfo meterReadInfo =  buildMeterReadInfo(readDay);
 
 
         //推送数据
         Query pushQuery = new Query();
-        pushQuery.addCriteria(Criteria.where("deviceId").is(waterMeter.getId()));
+        //pushQuery.addCriteria(Criteria.where("deviceId").is(waterMeter.getId()));
         Update push = new Update();
         push.push("data",meterReadInfo);
         long pushResult = mongoTemplate.updateMulti(pushQuery,push, Constants.METER_READ_RECORD_TABLE).getModifiedCount();
-        if(pushResult == 0){
-            //推送失败插入数据
-            MeterReadRecord meterReadRecord = buildMeterReadRecord(waterMeter,meterReadInfo);
-            mongoTemplate.insert(meterReadRecord,Constants.METER_READ_RECORD_TABLE);
-        }
-        log.info("end saveReadingData");
+        log.info("end saveReadingData pushResult = {}",pushResult);
     }
 
     private MeterReadRecord.MeterReadInfo buildMeterReadInfo(int readDay) {
@@ -103,10 +163,7 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
         return readInfo;
     }
 
-    private MeterReadRecord buildMeterReadRecord(WaterMeterDto waterMeter,MeterReadRecord.MeterReadInfo meterReadInfo) {
-
-        List<MeterReadRecord.MeterReadInfo> list = new ArrayList<>();
-        list.add(meterReadInfo);
+    private MeterReadRecord buildMeterReadRecord(WaterMeterDto waterMeter) {
 
         MeterReadRecord readRecord = new MeterReadRecord();
         readRecord.setId(idWorker.nextId());
@@ -134,7 +191,7 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
         readRecord.setCreateTime(new Date());
         readRecord.setCreatorName(Constants.SYS_FLAG);
         readRecord.setUpdateTime(new Date());
-        readRecord.setData(list);
+
 
         return readRecord;
     }

zoniot-rmcp/zoniot-xxljob-client/src/main/resources/application-dev.properties

@@ -5,11 +5,11 @@ logging.path=./logs/zoniot-xxljob-client
 
 
 #xxl-job �������Ĺ��̵ĵ�ַ
-xxl.job.admin.addresses=http://127.0.0.1:8080/xxl-job-admin/
+xxl.job.admin.addresses=http://10.0.0.62:8080/xxl-job-admin/
 xxl.job.executor.appname=zoniot-xxljob-client
-xxl.job.executor.ip=127.0.0.1
+xxl.job.executor.ip=
 xxl.job.executor.port=9999
-xxl.job.executor.logpath=/opt/sit/meter-reading-job/xxl-job/jobhandler
+xxl.job.executor.logpath=/app/sit/uims/zoniot-xxljob-client/xxl-job-log
 xxl.job.accessToken=
 xxl.job.executor.logretentiondays=30
 
@@ -55,7 +55,7 @@ logging.level.org.springframework.data.mongodb.core=DEBUG
 spring.redis.host=10.0.0.63
 spring.redis.port=6379
 spring.redis.database=2
-spring.redis.timeout=36000
+spring.redis.timeout=500000
 
 # Lettuce
 spring.redis.lettuce.pool.max-active=8

zoniot-rmcp/zoniot-xxljob-client/src/main/resources/application-sit.properties

@@ -49,13 +49,13 @@ spring.jackson.time-zone=GMT+8
 
 #mongodb url
 spring.data.mongodb.uri=mongodb://10.0.0.63:27017/meter-reading-database
-logging.level.org.springframework.data.mongodb.core=DEBUG
+#logging.level.org.springframework.data.mongodb.core=DEBUG
 
 # redis
 spring.redis.host=10.0.0.63
 spring.redis.port=6379
 spring.redis.database=2
-spring.redis.timeout=36000
+spring.redis.timeout=500000
 
 # Lettuce
 spring.redis.lettuce.pool.max-active=8

zoniot-rmcp/zoniot-xxljob-client/src/main/resources/logback-spring.xml

@@ -15,10 +15,10 @@
         <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
             <!-- rollover daily -->
             <fileNamePattern>${LOG_PATH}/${APPLICATION_NAME}-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
-            <!-- each file should be at most 100MB, keep 60 days worth of history, but at most 20GB -->
-            <maxFileSize>20MB</maxFileSize>
+            <!-- each file should be at most 100MB, keep 60 days worth of history, but at most 10GB -->
+            <maxFileSize>100MB</maxFileSize>
             <maxHistory>60</maxHistory>
-            <totalSizeCap>20GB</totalSizeCap>
+            <totalSizeCap>100GB</totalSizeCap>
         </rollingPolicy>
 
     </appender>