es增量数据同步

ch-admin-api/src/main/java/com/wwdz/ch/admin/job/SynItemToEsJob.java

@@ -58,9 +58,9 @@ public class SynItemToEsJob {
     RedissonClient redissonClient;
 
     /**
-     * 隔一小时运行一次
+     * 隔2分钟运行一次
      */
-//    @Scheduled(fixedDelay = 1000 * 60 * 60)
+    @Scheduled(fixedDelay = 1000 * 60 * 2)
     public void execute() {
         RLock lock = redissonClient.getLock(QUANKU_SYN_TO_ES_TASK_KEY);
         if (!lock.tryLock()) {
@@ -78,7 +78,7 @@ public class SynItemToEsJob {
             logger.info("es中最新数据时间为 {}", DateUtils.toString(time));
             CoinRequestDto coinRequestDto = new CoinRequestDto();
             coinRequestDto.setPage(1);
-            coinRequestDto.setLimit(200);
+            coinRequestDto.setLimit(10);
             coinRequestDto.setStartUpdateTime(time);
             coinRequestDto.setSortColumn("update_time");
             List<Item> itemList = itemDao.findListByPage(coinRequestDto);
@@ -86,17 +86,9 @@ public class SynItemToEsJob {
             logger.info("待同步增量数据总量 : {}, 当前查询 {} 条数据进行同步", pageInfo.getTotal(), itemList.size());
             for (Item item : itemList) {
                 executorService.submit(() -> {
-                    if (itemEsDao.isExised(item.getId())) {
-                        //存在就修改
-                        itemEsDao.update(item);
-                    } else {
-                        itemEsDao.createSingle(item);
-                    }
+                    itemEsDao.upsert(item);
                 });
             }
-
-
-            itemEsDao.create(itemList);
             logger.info(">>>>>>>>>>>>>>>>>>>>>>> 增量数据同步ES成功, 执行完毕 <<<<<<<<<<<<<<<<<<<<<");
         } catch (Exception e) {
             logger.error("同步item增量数据到ES error {}", e);

ch-admin-api/src/test/java/com/wwdz/ch/admin/impl/ElasticsearchUtilTest.java

@@ -57,7 +57,7 @@ public class ElasticsearchUtilTest {
     public void find() {
         EsSearchRequestDto esSearchRequestDto = new EsSearchRequestDto();
      //   esSearchRequestDto.setContent("清代");
-        esSearchRequestDto.setIncludes(Arrays.asList("id", "source", "name","images","updateTime", "createTime"));
+     //   esSearchRequestDto.setIncludes(Arrays.asList("id", "source", "name","images","updateTime", "createTime"));
         esSearchRequestDto.setPage(1);
         esSearchRequestDto.setSize(20);
         EsPageInfo esPageInfo = itemEsDao.findList(esSearchRequestDto);

ch-admin-api/src/test/java/com/wwdz/ch/admin/impl/SynCoinJobTest.java

@@ -2,6 +2,8 @@ package com.wwdz.ch.admin.impl;
 
 import com.wwdz.ch.admin.job.SynCoinJob;
 import com.wwdz.ch.admin.job.SynItemToEsJob;
+import com.wwdz.ch.db.dao.ItemDao;
+import com.wwdz.ch.db.domain.Item;
 import com.wwdz.ch.db.es.ItemEsDao;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -25,6 +27,9 @@ public class SynCoinJobTest {
     @Autowired
     SynItemToEsJob synItemToEsJob;
 
+    @Autowired
+    ItemDao itemDao;
+
     @Test
     public void execute() {
         synCoinJob.execute();
@@ -42,6 +47,18 @@ public class SynCoinJobTest {
         itemEsDao.createIndex();
     }
 
+    @Test
+    public void upsert() throws Exception {
+        Item item = itemDao.findById(767537L);
+        itemEsDao.upsert(item);
+    }
+
+    @Test
+    public void updateDocumnet() throws Exception {
+        Item item = itemDao.findById(767540L);
+        itemEsDao.update(item);
+    }
+
 
     @Test
     public void synToEs() throws Exception {

ch-core/src/main/resources/application-prod.yml

@@ -169,7 +169,7 @@ spring:
       uris: 172.16.122.190:9200
       username: elastic
       password: changeme
-      index: quanku_new
+      index: quanku_prod
 server:
   servlet:
     session:

ch-dao/src/main/java/com/wwdz/ch/db/es/ItemEsDao.java

@@ -163,10 +163,12 @@ public class ItemEsDao {
 
     }
 
-
-    public void createSingle(Item item) {
+    /**
+     * 更新或创建
+     * @param item
+     */
+    public void upsert(Item item) {
         try {
-            // 创建文档
             IndexRequest indexRequest = new IndexRequest.Builder()
                     .index(INDEX)
                     .id(String.valueOf(item.getId()))
@@ -174,9 +176,13 @@ public class ItemEsDao {
                     .build();
             IndexResponse indexResponse = elasticsearchClient.index(indexRequest);
             logger.info("indexResponse ={}", indexResponse);
-            logger.info("ES新增数据成功 : {} ", item);
+            if (indexResponse.shards().successful().intValue() == 1 ) {
+                logger.info("ES upsert 数据成功 : {} ", item);
+            } else {
+                logger.error("ES upsert 数据失败 : {} ", item);
+            }
         } catch (IOException e) {
-            logger.info("ES新增数据 content :{}  error" , item, e);
+            logger.info("ES upsert 数据 content :{}  error" , item, e);
         }
     }