change to mysql

elastic-load-data/config/application-mysql.yml

@@ -0,0 +1,10 @@
+database:
+  #url: jdbc:mysql:replication://192.168.10.210:6446,192.168.10.210:6447/fmmp?serverTimezone=Asia/Shanghai
+  url: jdbc:mysql://192.168.10.210:6446/fmmp?serverTimezone=Asia/Shanghai
+  user: root
+  password: R00T@mysql
+elasticsearch:
+  host: 192.168.10.241:9200,192.168.10.242:9200,192.168.10.243:9200
+  schema: http
+  user: elastic
+  password: abcd!234

elastic-load-data/config/application.yml

@@ -1,3 +1,3 @@
 profile:
-  active: dev
+  active: mysql
 #  active: production

elastic-load-data/pom.xml

@@ -95,6 +95,12 @@
             <groupId>cglib</groupId>
             <artifactId>cglib</artifactId>
         </dependency>
+        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>8.0.18</version>
+        </dependency>
 
     </dependencies>
 

elastic-load-data/src/main/java/com/vcarecity/test/module/ServiceModule.java

@@ -2,7 +2,7 @@ package com.vcarecity.test.module;
 
 import com.google.inject.AbstractModule;
 import com.vcarecity.test.service.ObtainDataService;
-import com.vcarecity.test.service.oracle.OracleObtainDataServiceImpl;
+import com.vcarecity.test.service.mysql.MySQLObtainDataServiceImpl;
 
 /**
  * @author Kerry on 19/12/09
@@ -11,6 +11,7 @@ import com.vcarecity.test.service.oracle.OracleObtainDataServiceImpl;
 public class ServiceModule extends AbstractModule {
     @Override
     protected void configure() {
-        bind(ObtainDataService.class).to(OracleObtainDataServiceImpl.class);
+        //bind(ObtainDataService.class).to(OracleObtainDataServiceImpl.class);
+        bind(ObtainDataService.class).to(MySQLObtainDataServiceImpl.class);
     }
 }

elastic-load-data/src/main/java/com/vcarecity/test/service/mysql/MySQLObtainDataServiceImpl.java

@@ -0,0 +1,73 @@
+package com.vcarecity.test.service.mysql;
+
+import com.google.inject.Inject;
+import com.vcarecity.test.entity.UnitAgencyEntity;
+import com.vcarecity.test.service.ObtainDataService;
+import com.vcarecity.test.typehandler.TypeHandler;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.List;
+
+/**
+ * @author VcKerry on 12/18/19
+ */
+
+@Slf4j
+public class MySQLObtainDataServiceImpl implements ObtainDataService {
+
+    //language=mysql
+    private static final String OB_SQL = "SELECT TU.UNIT_ID,\n" +
+            "       TU.DTU_NO,\n" +
+            "       TU.PUSER_CODE,\n" +
+            "       TAG.AGENCY_ID,\n" +
+            "       TAG.AGENCY_PATH,\n" +
+            "       TAG.AGENCY_NAME,\n" +
+            "       TB.BUILDING_ID,\n" +
+            "       TB.NAME_OF_BUILDING,\n" +
+            "       TUT.UNIT_TYPE_ID,\n" +
+            "       TUT.UNIT_TYPE\n" +
+            "FROM T_UNIT TU\n" +
+            "         INNER JOIN T_AGENCY TAG ON TU.AGENCY_ID = TAG.AGENCY_ID AND TU.ISDELETED = 0\n" +
+            "         INNER JOIN T_BUILDING TB ON TB.BUILDING_ID = TU.BUILDING_ID\n" +
+            "         INNER JOIN T_UNIT_TYPE TUT ON TUT.UNIT_TYPE_ID = TU.UNIT_TYPE_ID  limit ?,?";
+
+    private final DataSource dataSource;
+    private final TypeHandler typeHandler;
+
+    @Inject
+    public MySQLObtainDataServiceImpl(DataSource dataSource,
+                                      TypeHandler typeHandler) {
+        this.dataSource = dataSource;
+        this.typeHandler = typeHandler;
+        logger.info("MySQLObtainDataServiceImpl. dataSource = {}", dataSource);
+    }
+
+    @Override
+    public List<UnitAgencyEntity> queryUnitAgencyDataByPage(int page, int size) {
+
+        logger.debug("SQL = {}. page = {}, size = {}", OB_SQL, page, size);
+
+        int begin = page * size;
+        int end =  size;
+
+        try (Connection connection = dataSource.getConnection();
+             final PreparedStatement ps = connection.prepareStatement(OB_SQL)) {
+            ps.setInt(1, begin);
+            ps.setInt(2, end);
+
+            try (ResultSet resultSet = ps.executeQuery()) {
+                return typeHandler.resultHandler(resultSet, UnitAgencyEntity.class);
+            } catch (IllegalAccessException | InstantiationException e) {
+                e.printStackTrace();
+            }
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+}

elastic-load-data/src/main/java/com/vcarecity/test/service/oracle/OracleObtainDataServiceImpl.java

@@ -79,10 +79,10 @@ public class OracleObtainDataServiceImpl implements ObtainDataService {
             "                TB.BUILDING_ID,\n" +
             "                TB.NAME_OF_BUILDING,\n" +
             "                TUT.UNIT_TYPE_ID,\n" +
-            "                TUT.UNIT_TYPE," +
+            "                TUT.UNIT_TYPE,\n" +
             "                ROWNUM RN\n" +
-            "         FROM T_AGENCY TAG\n" +
-            "                  INNER JOIN T_UNIT TU ON TU.AGENCY_ID = TAG.AGENCY_ID AND TU.ISDELETED = 0\n" +
+            "         FROM T_UNIT TU\n" +
+            "                  INNER JOIN T_AGENCY TAG ON TU.AGENCY_ID = TAG.AGENCY_ID AND TU.ISDELETED = 0\n" +
             "                  INNER JOIN T_BUILDING TB ON TB.BUILDING_ID = TU.BUILDING_ID\n" +
             "                  INNER JOIN T_UNIT_TYPE TUT ON TUT.UNIT_TYPE_ID = TU.UNIT_TYPE_ID\n" +
             "         where ROWNUM <= ?\n" +

elastic-publish-service/pom.xml

@@ -29,11 +29,18 @@
             <artifactId>mybatis-spring-boot-starter</artifactId>
             <version>2.1.1</version>
         </dependency>
+
         <dependency>
-            <groupId>oracle.jdbc</groupId>
-            <artifactId>ojdbc8</artifactId>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
         </dependency>
 
+
+        <!-- <dependency>
+             <groupId>oracle.jdbc</groupId>
+             <artifactId>ojdbc8</artifactId>
+         </dependency>-->
+
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
             <artifactId>elasticsearch-rest-high-level-client</artifactId>
@@ -53,6 +60,13 @@
             <scope>provided</scope>
         </dependency>
 
+        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <version>8.0.18</version>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>

elastic-publish-service/src/main/java/com/vcarecity/publish/config/RedisConfig.java

@@ -0,0 +1,28 @@
+package com.vcarecity.publish.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+/**
+ * @author VcKerry on 12/18/19
+ */
+
+@Configuration
+public class RedisConfig {
+
+    @Bean
+    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
+        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
+        redisTemplate.setConnectionFactory(connectionFactory);
+        redisTemplate.setKeySerializer(new StringRedisSerializer());
+        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
+        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
+        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
+        return redisTemplate;
+    }
+
+}

elastic-publish-service/src/main/java/com/vcarecity/publish/constants/CheckConstant.java

@@ -0,0 +1,9 @@
+package com.vcarecity.publish.constants;
+
+/**
+ * @author VcKerry on 12/18/19
+ */
+
+public class CheckConstant {
+    public static final long CHECK_TIME_INTERVAL = 1000 * 3 * 60;
+}

elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/impl/ElasticUpdateServiceImpl.java

@@ -62,7 +62,6 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
         params.put("dtuNo", "000000000001");
         params.put("puserCode", "/0/000000000001/1/12345678/");
 
-
         UpdateByQueryRequest request = new UpdateByQueryRequest("unit-agency");
         request.setQuery(QueryBuilders.termQuery("unitId", 12345678));
         request.setMaxDocs(1);
@@ -88,8 +87,15 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
                 updateList.add(uae);
             }
         }
+
+        logger.info("key = {}, update = {}, delete = {}", key, updateList.size(), deleteId.size());
+        logger.debug("delete; {} : {}", key, deleteId);
+
         // 先删除
-        deleteByQuery(key, deleteId);
+        if (!deleteId.isEmpty()) {
+            this.deleteByQuery(key, deleteId);
+        }
+
         if (updateList.isEmpty()) {
             return;
         }
@@ -97,7 +103,7 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
         List<UnitAgencyMergeEntity> addData = updateDataByQuery(key, updateList);
 
         // 添加数据
-        addElasticData(addData);
+        this.addElasticData(addData);
     }
 
 
@@ -116,19 +122,15 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
     @Override
     public List<UnitAgencyMergeEntity> updateDataByQuery(String key, List<UnitAgencyMergeEntity> data) throws IOException {
         List<UnitAgencyMergeEntity> addList = new ArrayList<>();
-
+        final BeanCopier beanCopier = BeanCopier.create(UnitAgencyMergeEntity.class, UnitAgencyDetailEntity.class, false);
         for (UnitAgencyMergeEntity item : data) {
 
-            Map<String, Object> params = objectMapper.convertValue(item, new TypeReference<Map<String, Object>>() {
-                @Override
-                public Type getType() {
-                    return super.getType();
-                }
-            });
+            Map<String, Object> params = createElasticDataMap(beanCopier, item);
+            Object value = params.get(key);
+            params.remove(key);
 
             UpdateByQueryRequest request = new UpdateByQueryRequest(INDEX);
-            request.setQuery(QueryBuilders.termQuery(key, params.get(key)));
-
+            request.setQuery(QueryBuilders.termQuery(key, value));
             request.setScript(new Script(ScriptType.INLINE, "painless", getPainlessCode(params), params));
 
             BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
@@ -153,32 +155,14 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
         List<Map<String, Object>> saveResult = new ArrayList<>(data.size());
 
         for (UnitAgencyMergeEntity item : data) {
-            List<Integer> agencyIds = AgencyPathUtil.splitAgencyId(item.getAgencyPath());
-            List<String> agencyPathDetail = agencyService.getAgencyPathDetail(agencyIds);
-
-            UnitAgencyDetailEntity target = new UnitAgencyDetailEntity();
-            beanCopier.copy(item, target, null);
-
-            target.setAgencyDetail(String.join("", agencyPathDetail));
-
-            Map<String, Object> map = objectMapper.convertValue(target, new TypeReference<Map<String, Object>>() {
-                @Override
-                public Type getType() {
-                    return super.getType();
-                }
-            });
-            final Map<String, Integer> queryMap = AgencyPathUtil.agencyPathQueryMap(agencyIds);
-            map.putAll(queryMap);
-
+            Map<String, Object> map = createElasticDataMap(beanCopier, item);
             saveResult.add(map);
-
         }
         if (saveResult.isEmpty()) {
             return;
         }
         try {
             BulkRequest bulkRequest = new BulkRequest();
-
             for (Map<String, Object> val : saveResult) {
                 IndexRequest indexRequest = createIndexRequest(INDEX, val);
                 bulkRequest.add(indexRequest);
@@ -188,7 +172,23 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
         } catch (IOException e) {
             e.printStackTrace();
         }
+    }
 
+    private Map<String, Object> createElasticDataMap(BeanCopier beanCopier, UnitAgencyMergeEntity item) {
+        List<Integer> agencyIds = AgencyPathUtil.splitAgencyId(item.getAgencyPath());
+        List<String> agencyPathDetail = agencyService.getAgencyPathDetail(agencyIds);
+        UnitAgencyDetailEntity target = new UnitAgencyDetailEntity();
+        beanCopier.copy(item, target, null);
+        target.setAgencyDetail(String.join("", agencyPathDetail));
+        Map<String, Object> map = objectMapper.convertValue(target, new TypeReference<Map<String, Object>>() {
+            @Override
+            public Type getType() {
+                return super.getType();
+            }
+        });
+        final Map<String, Integer> queryMap = AgencyPathUtil.agencyPathQueryMap(agencyIds);
+        map.putAll(queryMap);
+        return map;
     }
 
 

elastic-publish-service/src/main/java/com/vcarecity/publish/service/impl/CheckTimeServiceImpl.java

@@ -1,14 +1,14 @@
 package com.vcarecity.publish.service.impl;
 
+import com.vcarecity.publish.constants.CheckConstant;
 import com.vcarecity.publish.service.CheckTimeService;
 import com.vcarecity.publish.util.DateUtil;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
 import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.Date;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * @author VcKerry on 12/17/19
@@ -17,22 +17,35 @@ import java.util.concurrent.ConcurrentHashMap;
 @Service
 public class CheckTimeServiceImpl implements CheckTimeService {
 
-    private final Map<String, Long> timeCache = new ConcurrentHashMap<>(4);
 
+    public static final String REDIS_KEY_PREFIX = "elastic:";
+
+    private final RedisTemplate<String, Object> redisTemplate;
+
+
+    public CheckTimeServiceImpl(RedisTemplate<String, Object> redisTemplate) {
+        this.redisTemplate = redisTemplate;
+    }
 
     @Override
     public String getLastDateTime(String key) {
+        String redisKey = REDIS_KEY_PREFIX + key;
 
-        Long aLong = timeCache.get(key);
-        if (aLong == null) {
-            aLong = System.currentTimeMillis();
-            timeCache.put(key, aLong);
+        Object value = redisTemplate.opsForValue().get(redisKey);
+
+        if (value != null) {
+            LocalDateTime.ofInstant(new Date(Long.parseLong(value.toString())).toInstant(), ZoneId.systemDefault()).format(DateUtil.DATE_TIME_FORMATTER);
         }
-        return LocalDateTime.ofInstant(new Date(aLong).toInstant(), ZoneId.systemDefault()).format(DateUtil.DATE_TIME_FORMATTER);
+
+        long last = System.currentTimeMillis() - CheckConstant.CHECK_TIME_INTERVAL;
+        redisTemplate.opsForValue().set(redisKey, last);
+
+        return LocalDateTime.ofInstant(new Date(last).toInstant(), ZoneId.systemDefault()).format(DateUtil.DATE_TIME_FORMATTER);
     }
 
     @Override
     public void setLastDateTime(String key, long lastDateTime) {
-        timeCache.put(key, lastDateTime);
+        String redisKey = REDIS_KEY_PREFIX + key;
+        redisTemplate.opsForValue().set(redisKey, lastDateTime);
     }
 }

elastic-publish-service/src/main/java/com/vcarecity/publish/task/CheckUpdateTask.java

@@ -6,6 +6,7 @@ import com.vcarecity.publish.mapper.UnitMapper;
 import com.vcarecity.publish.service.CheckTimeService;
 import com.vcarecity.publish.elastic.service.ElasticUpdateService;
 import com.vcarecity.publish.util.DateUtil;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 import java.time.LocalDateTime;
@@ -17,6 +18,7 @@ import java.util.List;
  * @author Kerry on 19/12/11
  */
 
+@Slf4j
 @Component
 public class CheckUpdateTask {
 
@@ -36,22 +38,28 @@ public class CheckUpdateTask {
         this.unitMapper = unitMapper;
     }
 
-
-    public void checkTask() {
+    public void checkUpdateTask() {
 
         // 当前更新时间
         long curCheckTime = System.currentTimeMillis();
         String toDateTime = LocalDateTime.ofInstant(new Date(curCheckTime).toInstant(), ZoneId.systemDefault()).format(DateUtil.DATE_TIME_FORMATTER);
 
+
         List<UnitAgencyMergeEntity> updateUaList;
         String fromDateTime;
 
         // 获取上一次的时间
         fromDateTime = checkTimeService.getLastDateTime("T_UNIT");
+
+        logger.info("start checkUpdateTask on T_UNIT from:{}. to:{}", fromDateTime, toDateTime);
+
         try {
             updateUaList = unitMapper.findUpdateUaList(fromDateTime, toDateTime);
+
+            logger.debug("T_UNIT update list size = {}", updateUaList.size());
+
             if (!updateUaList.isEmpty()) {
-                elasticUpdateService.updateUnit(updateUaList);
+                elasticUpdateService.handlerDataChange("unitId", updateUaList);
             }
             checkTimeService.setLastDateTime("T_UNIT", curCheckTime);
             updateUaList.clear();
@@ -59,12 +67,17 @@ public class CheckUpdateTask {
             e.printStackTrace();
         }
 
-
         fromDateTime = checkTimeService.getLastDateTime("T_AGENCY");
+
+        logger.info("start checkUpdateTask on T_AGENCY from:{}. to:{}", fromDateTime, toDateTime);
+
         try {
             updateUaList = agencyMapper.findUpdateUaList(fromDateTime, toDateTime);
+
+            logger.debug("T_AGENCY update list size = {}", updateUaList.size());
+
             if (!updateUaList.isEmpty()) {
-                elasticUpdateService.updateAgency(updateUaList);
+                elasticUpdateService.handlerDataChange("agencyId", updateUaList);
             }
             checkTimeService.setLastDateTime("T_AGENCY", curCheckTime);
             updateUaList.clear();

elastic-publish-service/src/main/resources/application-dev.yml

@@ -2,9 +2,9 @@ server:
   port: 7080
 spring:
   datasource:
-    url: jdbc:oracle:thin:@192.168.10.112:1521:orcl
-    username: C##FMMP
-    password: vcare~1(^_^)
+    url: jdbc:mysql:replication://192.168.10.210:6446,192.168.10.210:6447/fmmp?serverTimezone=Asia/Shanghai
+    username: root
+    password: R00T@mysql
   elasticsearch:
     rest:
       uris:
@@ -13,8 +13,11 @@ spring:
         - http://192.168.10.243:9200
       username: elastic
       password: abcd!234
-
-debug: true
+  redis:
+    database: 2
+    host: 192.168.10.112
+    port: 6379
+    password: admin~1(^-^)
 logging:
   level:
     com:
@@ -22,3 +25,5 @@ logging:
 mybatis:
   type-aliases-package: com.vcarecity.publish.entity
   mapper-locations: classpath:mapper/*.xml
+
+debug: true

elastic-publish-service/src/main/resources/mapper/AgencyMapper.xml

@@ -37,12 +37,10 @@
                TB.BUILDING_ID,
                TB.NAME_OF_BUILDING,
                TUT.UNIT_TYPE_ID,
-               TUT.UNIT_TYPE,
-               ROWNUM RN
+               TUT.UNIT_TYPE
         FROM (select *
               from T_AGENCY I_TAG
-              where UPDATESTAMP between to_date('${fromDate}', 'YYYY-MM-DD HH24:MI:SS')
-                        and to_date('${toDate}', 'YYYY-MM-DD HH24:MI:SS')
+              where UPDATESTAMP between '${fromDate}' and '${toDate}'
              ) TAG
                  INNER JOIN T_UNIT TU ON TU.AGENCY_ID = TAG.AGENCY_ID
                  INNER JOIN T_BUILDING TB ON TB.BUILDING_ID = TU.BUILDING_ID

elastic-publish-service/src/main/resources/mapper/UnitMapper.xml

@@ -35,12 +35,10 @@
                TB.BUILDING_ID,
                TB.NAME_OF_BUILDING,
                TUT.UNIT_TYPE_ID,
-               TUT.UNIT_TYPE,
-               ROWNUM RN
+               TUT.UNIT_TYPE
         FROM (select *
               from T_UNIT I_TU
-              where UPDATESTAMP between to_date('${fromDate}', 'YYYY-MM-DD HH24:MI:SS')
-                        and to_date('${toDate}', 'YYYY-MM-DD HH24:MI:SS')
+              where UPDATESTAMP between '${fromDate}' and '${toDate}'
              ) TU
                  INNER JOIN T_AGENCY TAG ON TU.AGENCY_ID = TAG.AGENCY_ID
                  INNER JOIN T_BUILDING TB ON TB.BUILDING_ID = TU.BUILDING_ID

elastic-publish-service/src/test/java/com/vcarecity/publish/mapper/UnitMapperTest.java

@@ -1,10 +1,12 @@
 package com.vcarecity.publish.mapper;
 
 import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
+import com.vcarecity.publish.util.DateUtil;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 
+import java.time.LocalDateTime;
 import java.util.List;
 
 @SpringBootTest
@@ -22,12 +24,18 @@ class UnitMapperTest {
 
     @Test
     public void findUpdateUaList() {
-        String from = "2018-12-23 15:00:00";
-        String to = "2018-12-24 15:00:00";
+
+        LocalDateTime now = LocalDateTime.now();
+        LocalDateTime last = now.minusHours(1L);
+
+        String from = last.format(DateUtil.DATE_TIME_FORMATTER);
+        String to = now.format(DateUtil.DATE_TIME_FORMATTER);
         List<UnitAgencyMergeEntity> updateUaList = unitMapper.findUpdateUaList(from, to);
 
+        System.out.println(updateUaList.size());
+
         for (UnitAgencyMergeEntity unitAgencyMergeEntity : updateUaList) {
-            System.out.println(unitAgencyMergeEntity);
+            // System.out.println(unitAgencyMergeEntity);
         }
     }
 }

elastic-publish-service/src/test/java/com/vcarecity/publish/task/CheckUpdateTaskTest.java

@@ -0,0 +1,16 @@
+package com.vcarecity.publish.task;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class CheckUpdateTaskTest {
+    @Autowired
+    private CheckUpdateTask checkUpdateTask;
+
+    @Test
+    void checkUpdateTask() {
+        checkUpdateTask.checkUpdateTask();
+    }
+}