Merge remote-tracking branch 'origin/20200908' into 20200908

meter-reading-common/src/main/java/com/huaxu/zoniot/entity/MeterReadRecord.java

@@ -125,6 +125,12 @@ public class MeterReadRecord implements Serializable {
     */
     private String readData;
 
+    /**
+     * 阀门状态
+     * 0：关 1：开 2：无阀 3：异常
+     */
+    private Integer valveState ;
+
     /**
     * 最近有效数据
     */

meter-reading-common/src/main/java/com/huaxu/zoniot/entity/WaterMeter.java

@@ -84,6 +84,10 @@ public class WaterMeter {
      */
     private Integer deviceTypeId;
 
+    /**
+     * 是否支持阀控,0:不支持,1:支持
+     */
+    private Integer isValve;
 
     /**
      * 阀门测点

meter-reading-common/src/main/java/com/huaxu/zoniot/service/impl/MeterReadRecordServiceImpl.java

@@ -216,11 +216,71 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
         meterReadRecord.setCommunityName(waterMeter.getCommunityName());
         meterReadRecord.setChannelNumberId(waterMeter.getChannelNumberId());
         meterReadRecord.setChannelName(waterMeter.getChannelName());
-
+        /***************************增加阀门状态**********************************/
+        meterReadRecord.setValveState(getValveState(waterMeter));
         return meterReadRecord ;
     }
 
 
+    /**
+     * 获取上一次阀门状态
+     * 平台： 0：关 1：开 2：无阀 3：异常 4:未知
+     * NB：0开，1关，2故障
+     * Lora：00开，01关，11异常
+     * @param waterMeter
+     * @return
+     */
+    protected  Integer getValveState(WaterMeter waterMeter){
+        Integer valveState = null ;
+        // 表是否支持阀控
+        Integer isValve = waterMeter.getIsValve();
+        if(isValve == 0){
+            // 不支持阀控,则状态为无阀
+            valveState = 2 ;
+        }
+        else{
+            // 支持阀控,则从最后一次上报数据中获取阀门状态
+            String lastValveState = getValveStateFromCache(waterMeter);
+            if(StringUtils.isNotBlank(lastValveState)){
+                if("00".equals(lastValveState) || "0".equals(lastValveState)){
+                    valveState = 1 ;
+                }
+                else if("01".equals(lastValveState) || "1".equals(lastValveState)){
+                    valveState = 0 ;
+                }
+                else if("11".equals(lastValveState) || "2".equals(lastValveState)){
+                    valveState = 3 ;
+                }
+                else{ // 未明确的阀门状态
+                    valveState = 4 ;
+                }
+            }
+            else{
+                valveState = 4 ;
+            }
+        }
+        return valveState ;
+    }
+
+    protected String getValveStateFromCache(WaterMeter waterMeter){
+        // 默认水表止度
+        String lastValveState = "" ;
+        Map<String, MeasuringData> meterLastData = getMeterLastDataFromCache(waterMeter.getDeviceId());
+        if(meterLastData != null){
+            MeasuringData measuringData = meterLastData.get(waterMeter.getValveMeasuringCode());
+            if(measuringData == null){
+                log.error("Device measuring code get null value ? " +
+                                "device Id ={},measuring code ={} ,meterLastData = {}",waterMeter.getDeviceId(),
+                        waterMeter.getReadingMeasuringCode(),JSON.toJSONString(meterLastData));
+            }
+            else{
+                if(StringUtils.isNotBlank(measuringData.getMeasuringVaule())){
+                    lastValveState = measuringData.getMeasuringVaule();
+                }
+            }
+        }
+        return lastValveState ;
+    }
     /**
      * 获取水表最后止度,先从缓存中获取止度,缓存无数据则从数据库中获取
      * @param waterMeter
@@ -384,6 +444,7 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
                 }
             }
             String currentReading = getReadingFromMeterData(waterMeter, measuringData);
+            Integer valveState = getValveStateFromMeterData(waterMeter, measuringData);
             if(StringUtils.isBlank(currentReading)){
                 log.error("表{}数据异常,数据无最新止度,MeasuringData = {}",
                         waterMeter.getDeviceId(),JSON.toJSONString(measuringData));
@@ -398,6 +459,7 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
             update.setLastCost(todayCost);
             update.setReadData(currentReading);
             update.setLastValid(currentReading);
+            update.setValveState(valveState);
             update.setReadStatus(Constants.READ);
             update.setDateUpdate(new Date());
             update.setReadTime(sendTime);
@@ -425,6 +487,7 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
                 param.set("lastCost",todayCost);
                 param.set("readData",currentReading);
                 param.set("lastValid",currentReading);
+                param.set("valveState",valveState);
                 param.set("readStatus",Constants.READ);
                 param.set("readTime",sendTime);
                 param.set("dateUpdate",new Date());
@@ -479,6 +542,41 @@ public class MeterReadRecordServiceImpl implements MeterReadRecordService {
         meterReading(meter,measuringData,readDay);
     }
 
+    protected  Integer getValveStateFromMeterData(WaterMeter waterMeter,Map<String, MeasuringData> measuringData){
+        Integer valveState = null ;
+        // 表是否支持阀控
+        Integer isValve = waterMeter.getIsValve();
+        if(isValve == 0){
+            // 不支持阀控,则状态为无阀
+            valveState = 2 ;
+        }
+        else{
+            MeasuringData data = measuringData.get(waterMeter.getValveMeasuringCode());
+            if(data == null || StringUtils.isBlank(data.getMeasuringVaule())){
+                log.warn("Upload Data Not Contain Valve Data , MeasuringCode = {},MeasuringData = {}",
+                        waterMeter.getValveMeasuringCode(),
+                        JSON.toJSONString(measuringData)
+                        );
+                valveState = 4 ;
+            }
+            else{
+               String  valveStateData = data.getMeasuringVaule();
+                if("00".equals(valveStateData) || "0".equals(valveStateData)){
+                    valveState = 1 ;
+                }
+                else if("01".equals(valveStateData) || "1".equals(valveStateData)){
+                    valveState = 0 ;
+                }
+                else if("11".equals(valveStateData) || "2".equals(valveStateData)){
+                    valveState = 3 ;
+                }
+                else{ // 未明确的阀门状态
+                    valveState = 4 ;
+                }
+            }
+        }
+        return valveState ;
+    }
 
     /**
      * 从上报数据中获取当前读表数

meter-reading-common/src/main/resources/mapper/WaterMeterMapper.xml

@@ -38,7 +38,7 @@
       <result column="device_type_name" jdbcType="VARCHAR" property="deviceTypeName"/>
       <result column="manufacturer_id" jdbcType="INTEGER" property="manufacturerId" />
       <result column="manufacturer_name" jdbcType="VARCHAR" property="manufacturerName" />
-
+    <result column="is_valve" jdbcType="INTEGER" property="isValve" />
 
   </resultMap>
     <insert id="insert" keyProperty="waterMeter.id">
@@ -122,7 +122,8 @@
 	  m.`name` as manufacturer_name,
       cu.customer_name as customer_name,
       cn.channel_name as channel_name ,
-      wrd.channel_number_id as channel_number_id
+      wrd.channel_number_id as channel_number_id,
+      dt.is_valve as is_valve
   </sql>
   <sql id="Base_Meter_Query_Where">
       d.`status` = 1

meter-reading-data-migrate/pom.xml

@@ -21,14 +21,107 @@
     </properties>
     <dependencies>
         <dependency>
-            <groupId>com.huaxu.zoniot</groupId>
-            <artifactId>meter-reading-common</artifactId>
-            <version>1.0</version>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
         </dependency>
+        <!--支持aop注解功能-->
         <dependency>
             <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-web</artifactId>
+            <artifactId>spring-boot-starter-aop</artifactId>
+        </dependency>
+        <!--解决spring boot configuration annotation processor no found in classpath提示-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <!-- Redis -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
+        <!-- FastJSON -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.41</version>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <!--Common Lang-->
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-pool2</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>commons-collections</groupId>
+            <artifactId>commons-collections</artifactId>
+            <version>3.2.2</version>
+        </dependency>
+        <!-- 糊涂工具包 -->
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>5.5.4</version>
+        </dependency>
+        <!--druid数据源-->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>druid</artifactId>
+            <version>1.0.26</version>
+        </dependency>
+        <!--jdbc-->
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-jdbc</artifactId>
+            <version>5.1.3.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>8.0.11</version>
+        </dependency>
+        <!-- mybatis -->
+        <dependency>
+            <groupId>org.mybatis.spring.boot</groupId>
+            <artifactId>mybatis-spring-boot-starter</artifactId>
+            <version>2.1.3</version>
+        </dependency>
+        <!-- mongodb -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
+        <!-- spring batch -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-batch</artifactId>
+            <version>2.2.5.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>javax.validation</groupId>
+            <artifactId>validation-api</artifactId>
+        </dependency>
+        <!-- httpclient -->
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.6</version>
+        </dependency>
+        <!--
+        <dependency>
+            <groupId>com.huaxu.zoniot</groupId>
+            <artifactId>meter-reading-common</artifactId>
+            <version>1.0</version>
         </dependency>
+        -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
@@ -71,7 +164,7 @@
                         <manifest>
                             <addClasspath>true</addClasspath>
                             <classpathPrefix>lib/</classpathPrefix>
-                            <mainClass>com.huaxu.zoniot.MeterReadingJobApplication</mainClass>
+                            <mainClass>com.huaxu.zoniot.DataMigrateApplication</mainClass>
                         </manifest>
                     </archive>
                 </configuration>

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/DataMigrateApplication.java

@@ -1,7 +1,8 @@
 package com.huaxu.zoniot;
 
+import com.huaxu.zoniot.config.DruidConfig;
 import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 
 /**
  * <p>数据迁移程序</p>
@@ -10,7 +11,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
  * @Date 2021/1/29 15:54
  * @Version 1.0
  */
-@SpringBootApplication
+@EnableConfigurationProperties(DruidConfig.class)
 public class DataMigrateApplication {
     public static void main(String[] args) {
         SpringApplication.run(DataMigrateApplication.class,args);

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/config/AliYunDataMigrateJobConfiguration.java

@@ -0,0 +1,24 @@
+package com.huaxu.zoniot.config;
+
+import org.springframework.batch.core.ExitStatus;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.StepExecutionListener;
+
+/**
+ * <p>阿里云MySQL数据迁移到娄山云MySQL中</p>
+ *
+ * @Author wilian.peng
+ * @Date 2021/3/16 16:40
+ * @Version 1.0
+ */
+public class AliYunDataMigrateJobConfiguration implements StepExecutionListener {
+    @Override
+    public void beforeStep(StepExecution stepExecution) {
+
+    }
+
+    @Override
+    public ExitStatus afterStep(StepExecution stepExecution) {
+        return null;
+    }
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/config/DruidConfig.java

@@ -0,0 +1,157 @@
+package com.huaxu.zoniot.config;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.druid.support.http.StatViewServlet;
+import com.alibaba.druid.support.http.WebStatFilter;
+
+import com.huaxu.zoniot.datasource.DynamicDataSource;
+import com.huaxu.zoniot.datasource.DynamicDataSourceContextHolder;
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.jdbc.DataSourceBuilder;
+import org.springframework.boot.web.servlet.FilterRegistrationBean;
+import org.springframework.boot.web.servlet.ServletRegistrationBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.PropertySource;
+
+import javax.sql.DataSource;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "spring.druid",ignoreInvalidFields = true)
+@PropertySource("classpath:application.properties")
+public class DruidConfig {
+
+    private String driverClassName;
+    //阿里云数据源
+    private String aLiYunUrl;
+    private String aLiYunUsername;
+    private String aLiYunPassword;
+    //娄山云数据源
+    private String louShanYunUrl;
+    private String louShanYunUsername;
+    private String louShanYunPassword;
+
+    private int maxActive;
+    private int minIdle;
+    private int initialSize;
+    private Long timeBetweenEvictionRunsMillis;
+    private Long minEvictableIdleTimeMillis;
+    private String validationQuery;
+    private boolean testWhileIdle;
+    private boolean testOnBorrow;
+    private boolean testOnReturn;
+    private boolean poolPreparedStatements;
+    private Integer maxPoolPreparedStatementPerConnectionSize;
+    private String filters;
+    private String connectionProperties;
+
+    //这里需要注意默认是读取的application.properties配置文件。
+    //如果你的配置文件不在默认文件中
+    //需要在类中引入配置文件例如：@PropertySource(value = "classpath:druid.properties")
+//    @Bean(destroyMethod = "close",initMethod = "init")，不使用多数据源时，可以使用@Bean，并注释掉切面与dynamicDataSource方法
+    public DataSource getALiYunDs(){
+        DruidDataSource druidDataSource = new DruidDataSource();
+        druidDataSource.setDriverClassName(driverClassName);
+        druidDataSource.setUrl(aLiYunUrl);
+        druidDataSource.setUsername(aLiYunUsername);
+        druidDataSource.setPassword(aLiYunPassword);
+        druidDataSource.setMaxActive(maxActive);
+        druidDataSource.setInitialSize(initialSize);
+        druidDataSource.setTimeBetweenConnectErrorMillis(timeBetweenEvictionRunsMillis);
+        druidDataSource.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
+        druidDataSource.setValidationQuery(validationQuery);
+        druidDataSource.setTestWhileIdle(testWhileIdle);
+        druidDataSource.setTestOnBorrow(testOnBorrow);
+        druidDataSource.setTestOnReturn(testOnReturn);
+        druidDataSource.setPoolPreparedStatements(poolPreparedStatements);
+        druidDataSource.setMaxPoolPreparedStatementPerConnectionSize(maxPoolPreparedStatementPerConnectionSize);
+        try {
+            druidDataSource.setFilters(filters);
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+        return druidDataSource;
+    }
+
+    public DataSource getLouShanYunDs(){
+        DruidDataSource druidDataSource = new DruidDataSource();
+        druidDataSource.setDriverClassName(driverClassName);
+        druidDataSource.setUrl(louShanYunUrl);
+        druidDataSource.setUsername(louShanYunUsername);
+        druidDataSource.setPassword(louShanYunPassword);
+        druidDataSource.setMaxActive(maxActive);
+        druidDataSource.setInitialSize(initialSize);
+        druidDataSource.setTimeBetweenConnectErrorMillis(timeBetweenEvictionRunsMillis);
+        druidDataSource.setValidationQuery(validationQuery);
+        druidDataSource.setTestWhileIdle(testWhileIdle);
+        druidDataSource.setTestOnBorrow(testOnBorrow);
+        druidDataSource.setTestOnReturn(testOnReturn);
+        druidDataSource.setPoolPreparedStatements(poolPreparedStatements);
+        druidDataSource.setMaxPoolPreparedStatementPerConnectionSize(maxPoolPreparedStatementPerConnectionSize);
+        try {
+            druidDataSource.setFilters(filters);
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+        return druidDataSource;
+    }
+
+    @Bean
+    public DataSource dynamicDataSource(){
+        Map<Object, Object> targetDataSources = new HashMap<>();
+        // 创建2个数据源
+        DataSource ds1 = getALiYunDs();
+        DataSource ds2 = getLouShanYunDs();
+        targetDataSources.put("aliyun", ds1);
+        targetDataSources.put("loushanyun", ds2);
+        DynamicDataSource dynamicDataSource = new DynamicDataSource();
+        // 添加到数据源对象中
+        dynamicDataSource.setTargetDataSources(targetDataSources);
+        // 默认使用ds2
+        dynamicDataSource.setDefaultTargetDataSource(ds2);
+        // 将这个2个数据源添加到自己创建的动态数据源上下文对象中
+        // 用于判断后续业务方法中使用的数据源是否存在，不存在则使用默认的
+        DynamicDataSourceContextHolder.dataSourceIds.add("aliyun");
+        DynamicDataSourceContextHolder.dataSourceIds.add("loushanyun");
+        return dynamicDataSource;
+    }
+
+    /**
+     * 配置访问druid监控
+     */
+    @Bean
+    public ServletRegistrationBean druidStateViewServlet() {
+        ServletRegistrationBean servletRegistrationBean = new ServletRegistrationBean(new StatViewServlet(), "/druid/*");
+        //初始化参数initParams
+        //添加白名单
+        servletRegistrationBean.addInitParameter("allow","");
+        //添加ip黑名单
+        //servletRegistrationBean.addInitParameter("deny","192.168.0.11");
+        //登录查看信息的账号密码
+        servletRegistrationBean.addInitParameter("loginUsername","admin");
+        servletRegistrationBean.addInitParameter("loginPassword","123");
+        //是否能够重置数据
+        servletRegistrationBean.addInitParameter("resetEnable","false");
+        return servletRegistrationBean;
+    }
+
+    /**
+     * 过滤不需要监控的后缀
+     */
+    @Bean
+    public FilterRegistrationBean druidStatFilter(){
+        FilterRegistrationBean filterRegistrationBean = new FilterRegistrationBean(new WebStatFilter());
+        //添加过滤规则
+        filterRegistrationBean.addUrlPatterns("/*");
+        //添加不需要忽略的格式信息
+        filterRegistrationBean.addInitParameter("exclusions","*.js,*.gif,*.jpg,*.png,*.css,*.ico,/druid/*");
+        return filterRegistrationBean;
+    }
+
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/config/MySqlDataMigrateJobConfiguration.java

@@ -1,12 +1,12 @@
 package com.huaxu.zoniot.config;
 
-import com.huaxu.zoniot.common.CommonFileItemReader;
-import com.huaxu.zoniot.common.Constants;
-import com.huaxu.zoniot.common.ServiceException;
-import com.huaxu.zoniot.dto.WaterMigrateFileItemDto;
-import com.huaxu.zoniot.entity.MySqlDeviceData;
-import com.huaxu.zoniot.entity.WaterMeter;
-import com.huaxu.zoniot.entity.WaterMigrateFileItem;
+//import com.huaxu.zoniot.common.CommonFileItemReader;
+//import com.huaxu.zoniot.common.Constants;
+//import com.huaxu.zoniot.common.ServiceException;
+//import com.huaxu.zoniot.dto.WaterMigrateFileItemDto;
+//import com.huaxu.zoniot.entity.MySqlDeviceData;
+//import com.huaxu.zoniot.entity.WaterMeter;
+//import com.huaxu.zoniot.entity.WaterMigrateFileItem;
 import com.huaxu.zoniot.processor.MySqlDeviceDataProcessor;
 import com.huaxu.zoniot.reader.MySqlDeviceDataReader;
 import com.huaxu.zoniot.writer.DeviceDataWriter;
@@ -69,7 +69,8 @@ public class MySqlDataMigrateJobConfiguration  implements StepExecutionListener
 
     @Bean
     public Step mySqlDataMigrateStep() {
-        return stepBuilderFactory.get("mySqlDataMigrateStep")
+        return  null ;
+        /*return stepBuilderFactory.get("mySqlDataMigrateStep")
                 .listener(this)
                 .<WaterMigrateFileItem, WaterMeter>chunk(chunkSize)
                 .reader(mySqlDeviceDataReader())
@@ -81,7 +82,7 @@ public class MySqlDataMigrateJobConfiguration  implements StepExecutionListener
                 //.listener(migrateJobSkipListener)
                 .taskExecutor(taskExecutor())
                 .throttleLimit(8)
-                .build();
+                .build();*/
     }
 
     @Bean

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/config/SnowflakeIdWorkerConfig.java

@@ -0,0 +1,23 @@
+package com.huaxu.zoniot.config;
+
+import com.huaxu.zoniot.utils.SnowflakeIdWorker;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * 雪花算法分布式Id生成配置
+ * @author pengdi
+ */
+@Configuration
+public class SnowflakeIdWorkerConfig {
+    @Value("${server.dataCenterId}")
+    private long dataCenterId;
+
+    @Value("${server.workId}")
+    private long workId;
+    @Bean
+    public SnowflakeIdWorker snowflakeIdWorker() {
+        return new SnowflakeIdWorker(workId, dataCenterId);
+    }
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/dao/MeterReadRecordMapper.java

@@ -0,0 +1,20 @@
+package com.huaxu.zoniot.dao;
+
+import com.huaxu.zoniot.entity.MeterReadRecord;
+import java.util.List;
+import org.apache.ibatis.annotations.Param;
+
+/**
+ * <p></p>
+ * @Author wilian.peng
+ * @Date 2021/3/18 11:09
+ * @Version 1.0
+ */
+public interface MeterReadRecordMapper {
+
+    int insert(MeterReadRecord record);
+
+    int insertSelective(MeterReadRecord record);
+
+    int batchInsert(@Param("list") List<MeterReadRecord> list);
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/datasource/DynamicDataSource.java

@@ -0,0 +1,35 @@
+package com.huaxu.zoniot.datasource;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+/**
+ * @author pengdi
+ */
+@Slf4j
+public class DynamicDataSource extends AbstractRoutingDataSource {
+    @Override
+    protected Object determineCurrentLookupKey() {
+        try {
+            //通过反射获取父类数据源Map
+            Field targetDataSources = this.getClass().getSuperclass().getDeclaredField("targetDataSources");
+            targetDataSources.setAccessible(true);
+            Map<Object, Object> value = null;
+            value = (Map<Object, Object>) targetDataSources.get(this);
+            String ds = DynamicDataSourceContextHolder.getDataSource();
+            if (ds != null) {
+                log.info("当前数据源：{},url:{}",ds,((DruidDataSource)value.get(ds)).getUrl());
+            }
+            return DynamicDataSourceContextHolder.getDataSource();
+        } catch (NoSuchFieldException e) {
+            e.printStackTrace();
+        }catch (IllegalAccessException e) {
+            e.printStackTrace();
+        }
+        return "loushanyun";
+    }
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/datasource/DynamicDataSourceAspect.java

@@ -0,0 +1,36 @@
+package com.huaxu.zoniot.datasource;
+
+import lombok.extern.slf4j.Slf4j;
+import org.aspectj.lang.JoinPoint;
+import org.aspectj.lang.annotation.After;
+import org.aspectj.lang.annotation.Aspect;
+import org.aspectj.lang.annotation.Before;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+
+@Aspect
+@Order(-1)
+@Slf4j
+@Component
+public class DynamicDataSourceAspect {
+    //拦截有TargetDataSource注解的方法
+    @Before("@annotation(dataSource)")
+    public void changeDataSource(JoinPoint joinPoint, TargetDataSource dataSource) {
+        String dsId = dataSource.name();
+        //如果存在指定的数据源，则放入ThreadLocal中
+        if (DynamicDataSourceContextHolder.containsDataSource(dsId)) {
+            log.error("正在使用{}数据源,->{}", dsId, joinPoint.getSignature());
+            DynamicDataSourceContextHolder.setDataSource(dsId);
+        } else {
+            log.info("{}不存在,使用默认数据源,->{}", dsId, joinPoint.getSignature());
+        }
+    }
+    //释放数据库链接资源
+    @After(value = "@annotation(dataSource)")
+    public void releaseLocal(JoinPoint joinPoint, TargetDataSource dataSource) {
+        if (DynamicDataSourceContextHolder.getDataSource() != null) {
+            DynamicDataSourceContextHolder.getContextHolder().remove();
+            log.info("释放ds：" + dataSource.name() + "的ThreadLocal绑定！！！");
+        }
+    }
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/datasource/DynamicDataSourceContextHolder.java

@@ -0,0 +1,28 @@
+package com.huaxu.zoniot.datasource;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class DynamicDataSourceContextHolder {
+
+    private static final ThreadLocal<String> contextHolder = new ThreadLocal<>();
+
+    public static List<String> dataSourceIds = new ArrayList<>();
+
+    public static ThreadLocal<String> getContextHolder() {
+        return contextHolder;
+    }
+
+    public static void setDataSource(String dataSource){
+        contextHolder.set(dataSource);
+    }
+
+    public static String getDataSource(){
+        return contextHolder.get();
+    }
+
+    public static boolean containsDataSource(String dataSourceId) {
+        return dataSourceIds.contains(dataSourceId);
+    }
+
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/datasource/TargetDataSource.java

@@ -0,0 +1,13 @@
+package com.huaxu.zoniot.datasource;
+
+import java.lang.annotation.*;
+
+/**
+ * @author pengdi
+ */
+@Target({ElementType.METHOD, ElementType.TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface TargetDataSource {
+    String name();
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/entity/DeviceData.java

@@ -0,0 +1,59 @@
+package com.huaxu.zoniot.entity;
+
+import lombok.Data;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.mongodb.core.mapping.Document;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+import java.util.Date;
+import java.util.Map;
+
+/**
+ * 
+ * @Description  
+ * <p>设备数据表，按照设备ID进行分片</p>
+ * sh.enableSharding("meter-reading-database")
+ * sh.shardCollection("meter-reading-database.sc_device_data",{"receiveDate":1,"deviceId":1});
+ * @author wilian.peng
+ * @date 2019年11月12日 下午10:05:10
+ */
+@Data
+@Document(collection = "sc_device_data")
+public class DeviceData implements Serializable{
+	private static final long serialVersionUID = 4357722209160768792L;
+	@Id
+	private Long id;
+
+	private Long deviceId ;
+
+	private String deviceNo;
+
+	private Integer siteId ;
+
+	private Integer manufacturerId;
+
+	private Integer channelId;
+
+	private Integer deviceTypeId ;
+
+	private Integer buildingId;
+
+	private Integer communityId;
+
+	private String location ;
+
+	private Integer receiveDate ;
+
+	private Date receiveTime;
+
+	private Integer status;
+
+	private Map<String,String> data;
+
+	private LocalDateTime dateCreate;
+
+	private LocalDateTime dateUpdate ;
+
+	private Integer customerId;
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/entity/MeterReadRecord.java

@@ -0,0 +1,149 @@
+package com.huaxu.zoniot.entity;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import lombok.Data;
+
+/**
+ * <p></p>
+ * @Author wilian.peng
+ * @Date 2021/3/18 11:09
+ * @Version 1.0
+ */
+@Data
+public class MeterReadRecord {
+    /**
+    * 读表日期
+    */
+    private Integer readDate;
+
+    /**
+    * 站点
+    */
+    private Integer siteId;
+
+    /**
+    * 场景
+    */
+    private Integer sysId;
+
+    /**
+    * 省
+    */
+    private Integer province;
+
+    /**
+    * 市
+    */
+    private Integer city;
+
+    /**
+    * 区
+    */
+    private Integer region;
+
+    /**
+    * 小区
+    */
+    private Integer community;
+
+    /**
+    * 客户
+    */
+    private Integer customerId;
+
+    /**
+    * 集中器
+    */
+    private Integer concentratorId;
+
+    /**
+    * 采集器
+    */
+    private Integer collectorId;
+
+    /**
+    * 建筑
+    */
+    private Integer buildingId;
+
+    /**
+    * 安装地址
+    */
+    private String location;
+
+    /**
+    * 设备类型
+    */
+    private Integer deviceTypeId;
+
+    /**
+    * 设备id
+    */
+    private Long deviceId;
+
+    /**
+    * 节点编号
+    */
+    private String deviceNo;
+
+    /**
+    * 电子号
+    */
+    private String meterNo;
+
+    /**
+    * 档案号
+    */
+    private String meterFileNo;
+
+    /**
+    * 读表时间
+    */
+    private Date readTime;
+
+    /**
+    * 读表状态
+    */
+    private String readStatus;
+
+    /**
+    * 读表数据
+    */
+    private String readData;
+
+    /**
+    * 最近有效数据
+    */
+    private String lastValid;
+
+    /**
+    * 距离上次的消耗
+    */
+    private BigDecimal lastCost;
+
+    /**
+    * 状态
+    */
+    private Integer status;
+
+    /**
+    * 创建时间
+    */
+    private Date dateCreate;
+
+    /**
+    * 更新时间
+    */
+    private Date dateUpdate;
+
+    /**
+    * 创建人
+    */
+    private String createBy;
+
+    /**
+    * 更新人
+    */
+    private String updateBy;
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/entity/WaterMeter.java

@@ -0,0 +1,174 @@
+package com.huaxu.zoniot.entity;
+
+import lombok.Data;
+
+import java.util.Date;
+
+/**
+ * <p>水表</p>
+ *
+ * @Author wilian.peng
+ * @Date 2020/12/22 17:06
+ * @Version 1.0
+ */
+@Data
+public class WaterMeter {
+
+    /**
+     * 设备id
+     */
+    private Long deviceId;
+
+    /**
+     * 水表类型,1:物联网表,2:远传表
+     */
+    private Integer waterMeterType ;
+
+    /**
+     * 站点
+     */
+    private Integer siteId;
+
+    /**
+     * 场景
+     */
+    private Integer sysId;
+
+    /**
+     * 省
+     */
+    private Integer province;
+
+    /**
+     * 市
+     */
+    private Integer city;
+
+    /**
+     * 区
+     */
+    private Integer region;
+
+    /**
+     * 小区
+     */
+    private Integer community;
+
+    /**
+     * 客户
+     */
+    private Integer customerId;
+
+    /**
+     * 集中器
+     */
+    private Integer concentratorId;
+
+    /**
+     * 采集器
+     */
+    private Integer collectorId;
+
+    /**
+     * 建筑
+     */
+    private Integer buildingId;
+
+    /**
+     * 安装地址
+     */
+    private String location;
+
+    /**
+     * 设备类型
+     */
+    private Integer deviceTypeId;
+
+
+    /**
+     * 阀门测点
+     */
+    private String valveMeasuringCode;
+
+
+    /**
+     * 止度测点
+     */
+    private String readingMeasuringCode;
+
+    /**
+     * 节点编号
+     */
+    private String deviceNo;
+
+    /**
+     * 电子号
+     */
+    private String meterNo;
+
+    /**
+     * 档案号
+     */
+    private String meterFileNo;
+
+    /**
+     * 状态
+     */
+    private Integer status;
+
+    /**
+     * 创建时间
+     */
+    private Date dateCreate;
+
+    /**
+     * 更新时间
+     */
+    private Date dateUpdate;
+
+    /**
+     * 创建人
+     */
+    private String createBy;
+
+    /**
+     * 更新人
+     */
+    private String updateBy;
+
+    /**
+     * 设备状态
+     */
+    private Integer deviceStatus ;
+
+    /**
+     * 注册状态
+     */
+    private Integer registerStatus ;
+    /***********************************以下字段将存入MongoDB中**************************************/
+    private String communityName;
+
+    private String buildingName ;
+
+    private String customerName ;
+
+    private String provinceName ;
+
+    private String cityName ;
+
+    private String regionName ;
+
+    private String deviceTypeName ;
+
+    private String concentratorNo ;
+
+    private String collectorNo ;
+
+    private Integer manufacturerId ;
+
+    private String manufacturerName ;
+
+    private Integer channelNumberId;
+
+    private String channelName ;
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/processor/MySqlDeviceDataProcessor.java

@@ -5,7 +5,6 @@ import com.huaxu.zoniot.entity.Device;
 import com.huaxu.zoniot.entity.DeviceData;
 import com.huaxu.zoniot.entity.MySqlDeviceData;
 import com.huaxu.zoniot.service.DeviceService;
-import com.huaxu.zoniot.service.WaterMeterService;
 import com.huaxu.zoniot.utils.SnowflakeIdWorker;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/repository/DeviceDataRepository.java

@@ -0,0 +1,14 @@
+package com.huaxu.zoniot.repository;
+
+import com.huaxu.zoniot.entity.DeviceData;
+import org.springframework.data.mongodb.repository.MongoRepository;
+
+/**
+ * <p>设备数据MongoDB仓库类</p>
+ *
+ * @Author wilian.peng
+ * @Date 2021/1/7 11:41
+ * @Version 1.0
+ */
+public interface DeviceDataRepository extends MongoRepository<DeviceData,Long> {
+}

meter-reading-data-migrate/src/main/java/com/huaxu/zoniot/utils/SnowflakeIdWorker.java

@@ -0,0 +1,182 @@
+package com.huaxu.zoniot.utils;
+
+/**
+ * Twitter_Snowflake<br>
+ * SnowFlake的结构如下(每部分用-分开):<br>
+ * 0 - 0000000000 0000000000 0000000000 0000000000 0 - 00000 - 00000 - 000000000000 <br>
+ * 1位标识，由于long基本类型在Java中是带符号的，最高位是符号位，正数是0，负数是1，所以id一般是正数，最高位是0<br>
+ * 41位时间截(毫秒级)，注意，41位时间截不是存储当前时间的时间截，而是存储时间截的差值（当前时间截 - 开始时间截)
+ * 得到的值），这里的的开始时间截，一般是我们的id生成器开始使用的时间，由我们程序来指定的（如下下面程序IdWorker类的startTime属性）。41位的时间截，可以使用69年，年T = (1L << 41) / (1000L * 60 * 60 * 24 * 365) = 69<br>
+ * 10位的数据机器位，可以部署在1024个节点，包括5位datacenterId和5位workerId<br>
+ * 12位序列，毫秒内的计数，12位的计数顺序号支持每个节点每毫秒(同一机器，同一时间截)产生4096个ID序号<br>
+ * 加起来刚好64位，为一个Long型。<br>
+ * SnowFlake的优点是，整体上按照时间自增排序，并且整个分布式系统内不会产生ID碰撞(由数据中心ID和机器ID作区分)，并且效率较高，经测试，SnowFlake每秒能够产生26万ID左右。
+ */
+public class SnowflakeIdWorker {
+
+    // ==============================Fields===========================================
+    /**
+     * 开始时间截 (2015-01-01)
+     */
+    private final long twepoch = 1420041600000L;
+
+    /**
+     * 机器id所占的位数
+     */
+    private final long workerIdBits = 5L;
+
+    /**
+     * 数据标识id所占的位数
+     */
+    private final long datacenterIdBits = 5L;
+
+    /**
+     * 支持的最大机器id，结果是31 (这个移位算法可以很快的计算出几位二进制数所能表示的最大十进制数)
+     */
+    private final long maxWorkerId = -1L ^ (-1L << workerIdBits);
+
+    /**
+     * 支持的最大数据标识id，结果是31
+     */
+    private final long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);
+
+    /**
+     * 序列在id中占的位数
+     */
+    private final long sequenceBits = 12L;
+
+    /**
+     * 机器ID向左移12位
+     */
+    private final long workerIdShift = sequenceBits;
+
+    /**
+     * 数据标识id向左移17位(12+5)
+     */
+    private final long datacenterIdShift = sequenceBits + workerIdBits;
+
+    /**
+     * 时间截向左移22位(5+5+12)
+     */
+    private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;
+
+    /**
+     * 生成序列的掩码，这里为4095 (0b111111111111=0xfff=4095)
+     */
+    private final long sequenceMask = -1L ^ (-1L << sequenceBits);
+
+    /**
+     * 工作机器ID(0~31)
+     */
+    private long workerId;
+
+    /**
+     * 数据中心ID(0~31)
+     */
+    private long datacenterId;
+
+    /**
+     * 毫秒内序列(0~4095)
+     */
+    private long sequence = 0L;
+
+    /**
+     * 上次生成ID的时间截
+     */
+    private long lastTimestamp = -1L;
+
+    //==============================Constructors=====================================
+
+    /**
+     * 构造函数
+     *
+     * @param workerId     工作ID (0~31)
+     * @param datacenterId 数据中心ID (0~31)
+     */
+    public SnowflakeIdWorker(long workerId, long datacenterId) {
+        if (workerId > maxWorkerId || workerId < 0) {
+            throw new IllegalArgumentException(String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));
+        }
+        if (datacenterId > maxDatacenterId || datacenterId < 0) {
+            throw new IllegalArgumentException(String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
+        }
+        this.workerId = workerId;
+        this.datacenterId = datacenterId;
+    }
+
+    // ==============================Methods==========================================
+
+    /**
+     * 获得下一个ID (该方法是线程安全的)
+     *
+     * @return SnowflakeId
+     */
+    public synchronized long nextId() {
+        long timestamp = timeGen();
+
+        //如果当前时间小于上一次ID生成的时间戳，说明系统时钟回退过这个时候应当抛出异常
+        if (timestamp < lastTimestamp) {
+            throw new RuntimeException(
+                    String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
+        }
+
+        //如果是同一时间生成的，则进行毫秒内序列
+        if (lastTimestamp == timestamp) {
+            sequence = (sequence + 1) & sequenceMask;
+            //毫秒内序列溢出
+            if (sequence == 0) {
+                //阻塞到下一个毫秒,获得新的时间戳
+                timestamp = tilNextMillis(lastTimestamp);
+            }
+        }
+        //时间戳改变，毫秒内序列重置
+        else {
+            sequence = 0L;
+        }
+
+        //上次生成ID的时间截
+        lastTimestamp = timestamp;
+
+        //移位并通过或运算拼到一起组成64位的ID
+        return ((timestamp - twepoch) << timestampLeftShift) //
+                | (datacenterId << datacenterIdShift) //
+                | (workerId << workerIdShift) //
+                | sequence;
+    }
+
+    /**
+     * 阻塞到下一个毫秒，直到获得新的时间戳
+     *
+     * @param lastTimestamp 上次生成ID的时间截
+     * @return 当前时间戳
+     */
+    protected long tilNextMillis(long lastTimestamp) {
+        long timestamp = timeGen();
+        while (timestamp <= lastTimestamp) {
+            timestamp = timeGen();
+        }
+        return timestamp;
+    }
+
+    /**
+     * 返回以毫秒为单位的当前时间
+     *
+     * @return 当前时间(毫秒)
+     */
+    protected long timeGen() {
+        return System.currentTimeMillis();
+    }
+
+    //==============================Test=============================================
+
+    /**
+     * 测试
+     */
+    public static void main(String[] args) {
+        SnowflakeIdWorker idWorker = new SnowflakeIdWorker(0, 0);
+        for (int i = 0; i < 5; i++) {
+            long id = idWorker.nextId();
+            System.out.println(id);
+        }
+    }
+}

meter-reading-data-migrate/src/main/resources/ReadMe.md

@@ -0,0 +1,8 @@
+```
+本工程包含三个内容的数据迁移
+1.阿里云mysql数据迁移到娄山云mysql(抄表记录)
+使用
+2.阿里云mysql数据迁移到mongodb(设备历史数据)
+3.互联互通数据迁移到物联网平台(设备历史数据)
+```
+

meter-reading-data-migrate/src/main/resources/application-migrate-dev.properties

@@ -1 +1,100 @@
+#########################################基础配置#################################################
+server.workId=0
+server.dataCenterId=1
+#########################################数据源配置#################################################
+# 连接池的配置信息
+spring.druid.aLiYunUrl=jdbc:mysql://114.135.61.188:33306/smart_city_sit_6_10?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&zeroDateTimeBehavior=convertToNull
+spring.druid.aLiYunUsername=root
+spring.druid.aLiYunPassword=100Zone@123
+spring.druid.louShanYunUrl=jdbc:mysql://192.168.0.120:3306/smartcity?characterEncoding=utf8&useSSL=false&serverTimezone=Asia/Shanghai&zeroDateTimeBehavior=convertToNull
+spring.druid.louShanYunUsername=root
+spring.druid.louShanYunPassword=1234
+spring.druid.driver-class-name=com.mysql.cj.jdbc.Driver
+spring.druid.initialSize=5
+spring.druid.minIdle=2
+spring.druid.maxActive=5
+## 配置获取连接等待超时的时间
+spring.druid.maxWait=60000
+# 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位是毫秒
+spring.druid.timeBetweenEvictionRunsMillis=60000
+# 配置一个连接在池中最小生存的时间，单位是毫秒
+spring.druid.minEvictableIdleTimeMillis=300000
+spring.druid.validationQuery=SELECT 1 FROM DUAL
+spring.druid.testWhileIdle=true
+spring.druid.testOnBorrow=false
+spring.druid.testOnReturn=false
+spring.druid.poolPreparedStatements=true
+spring.druid.maxPoolPreparedStatementPerConnectionSize=20
+# 配置监控统计拦截的filters，去掉后监控界面sql无法统计，'wall'用于防火墙
+spring.druid.filters=stat,wall
+# 通过connectProperties属性来打开mergeSql功能；慢SQL记录
+spring.druid.connectionProperties=druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000
+#########################################MyBatis配置#################################################
+mybatis.mapper-locations=classpath*:mapper/*.xml
+mybatis.type-aliases-package=com.huaxu.zoniot.entity
+mybatis.configuration.map-underscore-to-camel-case=true
+mybatis.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl
+mybatis.configuration.use-column-label=true
+#########################################Redis配置#################################################
+spring.redis.host=114.135.61.188
+spring.redis.port=26379
+spring.redis.password=zoniot
+spring.redis.database=15
+spring.redis.lettuce.pool.max-active=8
+spring.redis.lettuce.pool.max-wait=100
+spring.redis.lettuce.pool.max-idle=8
+spring.redis.lettuce.pool.min-idle=0
+spring.redis.lettuce.shutdown-timeout=100
+######################################################MongoDB配置#####################################################
+#基础连接参数
+#要连接的数据库
+mongodb.database=meter-reading-database
+#用户名
+mongodb.username=
+#密码
+mongodb.password=
+#IP和端口（host:port），例如127.0.0.1:27017。集群模式用,分隔开，例如host1:port1,host2:port2
+mongodb.address=114.135.61.188:17017
+#设置认证数据库，如果有的话
+mongodb.authenticationDatabase=
+# 客户端连接池参数
+#客户端的标识，用于定位请求来源等，一般用程序名
+mongodb.clientName=${spring.application.name}
+#TCP（socket）连接超时时间，毫秒
+mongodb.connectionTimeoutMs=5000
+#TCP（socket）连接闲置时间，毫秒
+mongodb.maxConnectionIdleTimeMs=60000
+#TCP（socket）连接最多可以使用多久，毫秒
+mongodb.maxConnectionLifeTimeMs=300000
+#TCP（socket）读取超时时间，毫秒
+mongodb.readTimeoutMs=15000
+#当连接池无可用连接时客户端阻塞等待的最大时长，毫秒
+mongodb.maxWaitTimeMs=5000
+#心跳检测发送频率，毫秒
+mongodb.heartbeatFrequencyMs=20000
+#最小的心跳检测发送频率，毫秒
+mongodb.minHeartbeatFrequencyMs=8000
+#心跳检测连接超时时间，毫秒
+mongodb.heartbeatConnectionTimeoutMs=10000
+#心跳检测读取超时时间，毫秒
+mongodb.heartbeatReadTimeoutMs=15000
+#线程池允许的最大连接数
+mongodb.connectionsPerHost=20
+#线程池空闲时保持的最小连接数
+mongodb.minConnectionsPerHost=4
+#计算允许多少个线程阻塞等待时的乘数，算法：threadsAllowedToBlockForConnectionMultiplier*maxConnectionsPerHost
+mongodb.threadsAllowedToBlockForConnectionMultiplier=10
+#spring.data.mongodb.uri=mongodb://114.135.61.188:17017/meter-reading-database
+logging.level.org.springframework.data.mongodb.core=DEBUG
+######################################################SpringBatch配置#####################################################
+# 默认情况下，项目启动时就会自动执行配置好的批处理操作。这里将其设为不自动执行，后面我们通过手动触发执行批处理
+spring.batch.job.enabled=false
+# 项目启动时创建数据表（用于记录批处理执行状态）的 SQL 脚本，该脚本由Spring Batch提供
+spring.datasource.schema=classpath:/org/springframework/batch/core/schema-mysql.sql
+# 项目启动时执行建表 SQL
+spring.batch.initialize-schema=always
+# 每次执行的批大小
+spring.batch.chunk.size=100
+#########################################业务配置############################################
+#定义每个任务的数据集大小
 batch.data.chunk.size=2000

meter-reading-data-migrate/src/main/resources/mapper/MeterReadRecordMapper.xml

@@ -0,0 +1,256 @@
+<?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.huaxu.zoniot.dao.MeterReadRecordMapper">
+  <resultMap id="BaseResultMap" type="com.huaxu.zoniot.entity.MeterReadRecord">
+    <!--@mbg.generated-->
+    <!--@Table sc_meter_read_record-->
+    <result column="read_date" jdbcType="INTEGER" property="readDate" />
+    <result column="site_id" jdbcType="INTEGER" property="siteId" />
+    <result column="sys_id" jdbcType="INTEGER" property="sysId" />
+    <result column="province" jdbcType="INTEGER" property="province" />
+    <result column="city" jdbcType="INTEGER" property="city" />
+    <result column="region" jdbcType="INTEGER" property="region" />
+    <result column="community" jdbcType="INTEGER" property="community" />
+    <result column="customer_id" jdbcType="INTEGER" property="customerId" />
+    <result column="concentrator_id" jdbcType="INTEGER" property="concentratorId" />
+    <result column="collector_id" jdbcType="INTEGER" property="collectorId" />
+    <result column="building_id" jdbcType="INTEGER" property="buildingId" />
+    <result column="location" jdbcType="VARCHAR" property="location" />
+    <result column="device_type_id" jdbcType="INTEGER" property="deviceTypeId" />
+    <result column="device_id" jdbcType="BIGINT" property="deviceId" />
+    <result column="device_no" jdbcType="VARCHAR" property="deviceNo" />
+    <result column="meter_no" jdbcType="VARCHAR" property="meterNo" />
+    <result column="meter_file_no" jdbcType="VARCHAR" property="meterFileNo" />
+    <result column="read_time" jdbcType="TIMESTAMP" property="readTime" />
+    <result column="read_status" jdbcType="VARCHAR" property="readStatus" />
+    <result column="read_data" jdbcType="VARCHAR" property="readData" />
+    <result column="last_valid" jdbcType="VARCHAR" property="lastValid" />
+    <result column="last_cost" jdbcType="DECIMAL" property="lastCost" />
+    <result column="status" jdbcType="INTEGER" property="status" />
+    <result column="date_create" jdbcType="TIMESTAMP" property="dateCreate" />
+    <result column="date_update" jdbcType="TIMESTAMP" property="dateUpdate" />
+    <result column="create_by" jdbcType="VARCHAR" property="createBy" />
+    <result column="update_by" jdbcType="VARCHAR" property="updateBy" />
+  </resultMap>
+  <sql id="Base_Column_List">
+    <!--@mbg.generated-->
+    read_date, site_id, sys_id, province, city, region, community, customer_id, concentrator_id,
+    collector_id, building_id, `location`, device_type_id, device_id, device_no, meter_no, 
+    meter_file_no, read_time, read_status, read_data, last_valid, last_cost, `status`, 
+    date_create, date_update, create_by, update_by
+  </sql>
+  <insert id="insert" parameterType="com.huaxu.zoniot.entity.MeterReadRecord">
+    <!--@mbg.generated-->
+    insert into sc_meter_read_record (read_date, site_id,
+      sys_id, province, city, 
+      region, community, customer_id, 
+      concentrator_id, collector_id, building_id, 
+      `location`, device_type_id, device_id, 
+      device_no, meter_no, meter_file_no, 
+      read_time, read_status, read_data, 
+      last_valid, last_cost, `status`, 
+      date_create, date_update, create_by, 
+      update_by)
+    values (#{readDate,jdbcType=INTEGER}, #{siteId,jdbcType=INTEGER},
+      #{sysId,jdbcType=INTEGER}, #{province,jdbcType=INTEGER}, #{city,jdbcType=INTEGER}, 
+      #{region,jdbcType=INTEGER}, #{community,jdbcType=INTEGER}, #{customerId,jdbcType=INTEGER}, 
+      #{concentratorId,jdbcType=INTEGER}, #{collectorId,jdbcType=INTEGER}, #{buildingId,jdbcType=INTEGER}, 
+      #{location,jdbcType=VARCHAR}, #{deviceTypeId,jdbcType=INTEGER}, #{deviceId,jdbcType=BIGINT}, 
+      #{deviceNo,jdbcType=VARCHAR}, #{meterNo,jdbcType=VARCHAR}, #{meterFileNo,jdbcType=VARCHAR}, 
+      #{readTime,jdbcType=TIMESTAMP}, #{readStatus,jdbcType=VARCHAR}, #{readData,jdbcType=VARCHAR}, 
+      #{lastValid,jdbcType=VARCHAR}, #{lastCost,jdbcType=DECIMAL}, #{status,jdbcType=INTEGER}, 
+      #{dateCreate,jdbcType=TIMESTAMP}, #{dateUpdate,jdbcType=TIMESTAMP}, #{createBy,jdbcType=VARCHAR}, 
+      #{updateBy,jdbcType=VARCHAR})
+  </insert>
+  <insert id="insertSelective" parameterType="com.huaxu.zoniot.entity.MeterReadRecord">
+    <!--@mbg.generated-->
+    insert into sc_meter_read_record
+    <trim prefix="(" suffix=")" suffixOverrides=",">
+      <if test="readDate != null">
+        read_date,
+      </if>
+      <if test="siteId != null">
+        site_id,
+      </if>
+      <if test="sysId != null">
+        sys_id,
+      </if>
+      <if test="province != null">
+        province,
+      </if>
+      <if test="city != null">
+        city,
+      </if>
+      <if test="region != null">
+        region,
+      </if>
+      <if test="community != null">
+        community,
+      </if>
+      <if test="customerId != null">
+        customer_id,
+      </if>
+      <if test="concentratorId != null">
+        concentrator_id,
+      </if>
+      <if test="collectorId != null">
+        collector_id,
+      </if>
+      <if test="buildingId != null">
+        building_id,
+      </if>
+      <if test="location != null">
+        `location`,
+      </if>
+      <if test="deviceTypeId != null">
+        device_type_id,
+      </if>
+      <if test="deviceId != null">
+        device_id,
+      </if>
+      <if test="deviceNo != null">
+        device_no,
+      </if>
+      <if test="meterNo != null">
+        meter_no,
+      </if>
+      <if test="meterFileNo != null">
+        meter_file_no,
+      </if>
+      <if test="readTime != null">
+        read_time,
+      </if>
+      <if test="readStatus != null">
+        read_status,
+      </if>
+      <if test="readData != null">
+        read_data,
+      </if>
+      <if test="lastValid != null">
+        last_valid,
+      </if>
+      <if test="lastCost != null">
+        last_cost,
+      </if>
+      <if test="status != null">
+        `status`,
+      </if>
+      <if test="dateCreate != null">
+        date_create,
+      </if>
+      <if test="dateUpdate != null">
+        date_update,
+      </if>
+      <if test="createBy != null">
+        create_by,
+      </if>
+      <if test="updateBy != null">
+        update_by,
+      </if>
+    </trim>
+    <trim prefix="values (" suffix=")" suffixOverrides=",">
+      <if test="readDate != null">
+        #{readDate,jdbcType=INTEGER},
+      </if>
+      <if test="siteId != null">
+        #{siteId,jdbcType=INTEGER},
+      </if>
+      <if test="sysId != null">
+        #{sysId,jdbcType=INTEGER},
+      </if>
+      <if test="province != null">
+        #{province,jdbcType=INTEGER},
+      </if>
+      <if test="city != null">
+        #{city,jdbcType=INTEGER},
+      </if>
+      <if test="region != null">
+        #{region,jdbcType=INTEGER},
+      </if>
+      <if test="community != null">
+        #{community,jdbcType=INTEGER},
+      </if>
+      <if test="customerId != null">
+        #{customerId,jdbcType=INTEGER},
+      </if>
+      <if test="concentratorId != null">
+        #{concentratorId,jdbcType=INTEGER},
+      </if>
+      <if test="collectorId != null">
+        #{collectorId,jdbcType=INTEGER},
+      </if>
+      <if test="buildingId != null">
+        #{buildingId,jdbcType=INTEGER},
+      </if>
+      <if test="location != null">
+        #{location,jdbcType=VARCHAR},
+      </if>
+      <if test="deviceTypeId != null">
+        #{deviceTypeId,jdbcType=INTEGER},
+      </if>
+      <if test="deviceId != null">
+        #{deviceId,jdbcType=BIGINT},
+      </if>
+      <if test="deviceNo != null">
+        #{deviceNo,jdbcType=VARCHAR},
+      </if>
+      <if test="meterNo != null">
+        #{meterNo,jdbcType=VARCHAR},
+      </if>
+      <if test="meterFileNo != null">
+        #{meterFileNo,jdbcType=VARCHAR},
+      </if>
+      <if test="readTime != null">
+        #{readTime,jdbcType=TIMESTAMP},
+      </if>
+      <if test="readStatus != null">
+        #{readStatus,jdbcType=VARCHAR},
+      </if>
+      <if test="readData != null">
+        #{readData,jdbcType=VARCHAR},
+      </if>
+      <if test="lastValid != null">
+        #{lastValid,jdbcType=VARCHAR},
+      </if>
+      <if test="lastCost != null">
+        #{lastCost,jdbcType=DECIMAL},
+      </if>
+      <if test="status != null">
+        #{status,jdbcType=INTEGER},
+      </if>
+      <if test="dateCreate != null">
+        #{dateCreate,jdbcType=TIMESTAMP},
+      </if>
+      <if test="dateUpdate != null">
+        #{dateUpdate,jdbcType=TIMESTAMP},
+      </if>
+      <if test="createBy != null">
+        #{createBy,jdbcType=VARCHAR},
+      </if>
+      <if test="updateBy != null">
+        #{updateBy,jdbcType=VARCHAR},
+      </if>
+    </trim>
+  </insert>
+  <insert id="batchInsert" parameterType="map">
+    <!--@mbg.generated-->
+    insert into sc_meter_read_record
+    (read_date, site_id, sys_id, province, city, region, community, customer_id,
+      concentrator_id, collector_id, building_id, `location`, device_type_id, device_id, 
+      device_no, meter_no, meter_file_no, read_time, read_status, read_data, last_valid, 
+      last_cost, `status`, date_create, date_update, create_by, update_by)
+    values
+    <foreach collection="list" item="item" separator=",">
+      (#{item.readDate,jdbcType=INTEGER}, #{item.siteId,jdbcType=INTEGER},
+        #{item.sysId,jdbcType=INTEGER}, #{item.province,jdbcType=INTEGER}, #{item.city,jdbcType=INTEGER}, 
+        #{item.region,jdbcType=INTEGER}, #{item.community,jdbcType=INTEGER}, #{item.customerId,jdbcType=INTEGER}, 
+        #{item.concentratorId,jdbcType=INTEGER}, #{item.collectorId,jdbcType=INTEGER}, 
+        #{item.buildingId,jdbcType=INTEGER}, #{item.location,jdbcType=VARCHAR}, #{item.deviceTypeId,jdbcType=INTEGER}, 
+        #{item.deviceId,jdbcType=BIGINT}, #{item.deviceNo,jdbcType=VARCHAR}, #{item.meterNo,jdbcType=VARCHAR}, 
+        #{item.meterFileNo,jdbcType=VARCHAR}, #{item.readTime,jdbcType=TIMESTAMP}, #{item.readStatus,jdbcType=VARCHAR}, 
+        #{item.readData,jdbcType=VARCHAR}, #{item.lastValid,jdbcType=VARCHAR}, #{item.lastCost,jdbcType=DECIMAL}, 
+        #{item.status,jdbcType=INTEGER}, #{item.dateCreate,jdbcType=TIMESTAMP}, #{item.dateUpdate,jdbcType=TIMESTAMP}, 
+        #{item.createBy,jdbcType=VARCHAR}, #{item.updateBy,jdbcType=VARCHAR})
+    </foreach>
+  </insert>
+</mapper>