add elastic query delete update

common/src/main/java/com/vcarecity/elastic/util/SnowFlake.java

@@ -0,0 +1,119 @@
+package com.vcarecity.elastic.util;
+
+/**
+ * @author VcKerry
+ */
+public class SnowFlake {
+
+    /**
+     * 起始的时间戳:这个时间戳自己随意获取，比如自己代码的时间戳
+     */
+    private final static long START_STMP = 1543903501000L;
+
+    /**
+     * 每一部分占用的位数
+     */
+    private final static long SEQUENCE_BIT = 12; //序列号占用的位数
+    private final static long MACHINE_BIT = 5;  //机器标识占用的位数
+    private final static long DATACENTER_BIT = 5;//数据中心占用的位数
+
+    /**
+     * 每一部分的最大值：先进行左移运算，再同-1进行异或运算；异或：相同位置相同结果为0，不同结果为1
+     */
+    /**
+     * 用位运算计算出最大支持的数据中心数量：31
+     */
+    private final static long MAX_DATACENTER_NUM = -1L ^ (-1L << DATACENTER_BIT);
+
+    /**
+     * 用位运算计算出最大支持的机器数量：31
+     */
+    private final static long MAX_MACHINE_NUM = -1L ^ (-1L << MACHINE_BIT);
+
+    /**
+     * 用位运算计算出12位能存储的最大正整数：4095
+     */
+    private final static long MAX_SEQUENCE = -1L ^ (-1L << SEQUENCE_BIT);
+
+    /**
+     * 每一部分向左的位移
+     */
+
+    /**
+     * 机器标志较序列号的偏移量
+     */
+    private final static long MACHINE_LEFT = SEQUENCE_BIT;
+
+    /**
+     * 数据中心较机器标志的偏移量
+     */
+    private final static long DATACENTER_LEFT = SEQUENCE_BIT + MACHINE_BIT;
+
+    /**
+     * 时间戳较数据中心的偏移量
+     */
+    private final static long TIMESTMP_LEFT = DATACENTER_LEFT + DATACENTER_BIT;
+
+    private static long datacenterId;  //数据中心
+    private static long machineId;    //机器标识
+    private static long sequence = 0L; //序列号
+    private static long lastStmp = -1L;//上一次时间戳
+
+    /**
+     * 此处无参构造私有，同时没有给出有参构造，在于避免以下两点问题：
+     * 1、私有化避免了通过new的方式进行调用，主要是解决了在for循环中通过new的方式调用产生的id不一定唯一问题问题，因为用于			 记录上一次时间戳的lastStmp永远无法得到比对；
+     * 2、没有给出有参构造在第一点的基础上考虑了一套分布式系统产生的唯一序列号应该是基于相同的参数
+     */
+    private SnowFlake() {
+    }
+
+    /**
+     * 产生下一个ID
+     *
+     * @return
+     */
+    public static synchronized long nextId() {
+        /** 获取当前时间戳 */
+        long currStmp = getNewstmp();
+
+        /** 如果当前时间戳小于上次时间戳则抛出异常 */
+        if (currStmp < lastStmp) {
+            throw new RuntimeException("Clock moved backwards.  Refusing to generate id");
+        }
+        /** 相同毫秒内 */
+        if (currStmp == lastStmp) {
+            //相同毫秒内，序列号自增
+            sequence = (sequence + 1) & MAX_SEQUENCE;
+            //同一毫秒的序列数已经达到最大
+            if (sequence == 0L) {
+
+                /** 获取下一时间的时间戳并赋值给当前时间戳 */
+                currStmp = getNextMill();
+            }
+        } else {
+            //不同毫秒内，序列号置为0
+            sequence = 0L;
+        }
+        /** 当前时间戳存档记录，用于下次产生id时对比是否为相同时间戳 */
+        lastStmp = currStmp;
+
+
+        return (currStmp - START_STMP) << TIMESTMP_LEFT //时间戳部分
+                | datacenterId << DATACENTER_LEFT      //数据中心部分
+                | machineId << MACHINE_LEFT            //机器标识部分
+                | sequence;                            //序列号部分
+    }
+
+    private static long getNextMill() {
+        long mill = getNewstmp();
+        while (mill <= lastStmp) {
+            mill = getNewstmp();
+        }
+        return mill;
+    }
+
+    private static long getNewstmp() {
+        return System.currentTimeMillis();
+    }
+
+}

elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/ElasticIndexService.java

@@ -0,0 +1,14 @@
+package com.vcarecity.publish.elastic.service;
+
+/**
+ * @author kerryzhang on 2019/12/17
+ */
+
+public interface ElasticIndexService {
+
+    void createIndex(String ...indices);
+
+    void deleteIndex(String... indices);
+
+
+}

elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/ElasticQueryService.java

@@ -1,7 +1,5 @@
 package com.vcarecity.publish.elastic.service;
 
-import java.util.List;
-
 /**
  * @author VcKerry on 12/17/19
  */
@@ -9,5 +7,4 @@ import java.util.List;
 public interface ElasticQueryService {
 
 
-    List<Long> queryById();
 }

elastic-publish-service/src/main/java/com/vcarecity/publish/elastic/service/ElasticUpdateService.java

@@ -13,10 +13,28 @@ public interface ElasticUpdateService {
 
     void updateTest() throws IOException;
 
-    void updateUnit(List<UnitAgencyMergeEntity> list) throws IOException;
 
-
-    void updateAgency(List<UnitAgencyMergeEntity> list) throws IOException;
+    void handlerDataChange(String key, List<UnitAgencyMergeEntity> data) throws IOException;
+
+    /**
+     * 添加数据
+     *
+     * @param data
+     * @throws IOException
+     */
+    void addElasticData(List<UnitAgencyMergeEntity> data) throws IOException;
+
+    void deleteByQuery(String key, List<Long> ids) throws IOException;
+
+    /**
+     * update unit/agency... info,return incr data list
+     *
+     * @param key
+     * @param data
+     * @return
+     * @throws IOException
+     */
+    List<UnitAgencyMergeEntity> updateDataByQuery(String key, List<UnitAgencyMergeEntity> data) throws IOException;
 
 
 }

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

@@ -0,0 +1,23 @@
+package com.vcarecity.publish.elastic.service.impl;
+
+import com.vcarecity.publish.elastic.service.ElasticIndexService;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author kerryzhang on 2019/12/17
+ */
+
+@Service
+public class ElasticIndexServiceImpl implements ElasticIndexService {
+
+
+    @Override
+    public void createIndex(String... indices) {
+
+    }
+
+    @Override
+    public void deleteIndex(String... indices) {
+
+    }
+}

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

@@ -11,8 +11,5 @@ import java.util.List;
 
 @Service
 public class ElasticQueryServiceImpl implements ElasticQueryService {
-    @Override
-    public List<Long> queryById() {
-        return null;
-    }
+
 }

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

@@ -2,16 +2,26 @@ package com.vcarecity.publish.elastic.service.impl;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
+import com.vcarecity.elastic.util.AgencyPathUtil;
+import com.vcarecity.elastic.util.SnowFlake;
+import com.vcarecity.publish.elastic.entity.UnitAgencyDetailEntity;
 import com.vcarecity.publish.elastic.service.ElasticUpdateService;
+import com.vcarecity.publish.entity.UnitAgencyMergeEntity;
+import com.vcarecity.publish.sql.service.AgencyService;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 import org.elasticsearch.index.reindex.UpdateByQueryRequest;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptType;
+import org.springframework.cglib.beans.BeanCopier;
 import org.springframework.stereotype.Service;
 
 import java.io.IOException;
@@ -25,6 +35,7 @@ import java.util.Map;
  * @author Kerry on 12/17/19
  */
 
+@Slf4j
 @Service
 public class ElasticUpdateServiceImpl implements ElasticUpdateService {
 
@@ -32,11 +43,15 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
 
     private final RestHighLevelClient restHighLevelClient;
     private final ObjectMapper objectMapper;
+    private final AgencyService agencyService;
 
     public ElasticUpdateServiceImpl(RestHighLevelClient restHighLevelClient,
-                                    ObjectMapper objectMapper) {
+                                    ObjectMapper objectMapper,
+                                    AgencyService agencyService) {
         this.restHighLevelClient = restHighLevelClient;
         this.objectMapper = objectMapper;
+        this.agencyService = agencyService;
+
     }
 
 
@@ -62,85 +77,135 @@ public class ElasticUpdateServiceImpl implements ElasticUpdateService {
 
     }
 
-
-    private String getPainlessCode(Map<String, Object> params) {
-        String[] items = new String[params.size()];
-        int index = 0;
-        for (String key : params.keySet()) {
-            items[index++] = "ctx._source." + key + "=params." + key;
-        }
-        return String.join(";", items);
-    }
-
-
     @Override
-    public void updateUnit(List<UnitAgencyMergeEntity> list) throws IOException {
+    public void handlerDataChange(String key, List<UnitAgencyMergeEntity> data) throws IOException {
         List<UnitAgencyMergeEntity> updateList = new ArrayList<>();
         List<Long> deleteId = new ArrayList<>();
-        for (UnitAgencyMergeEntity uae : list) {
+        for (UnitAgencyMergeEntity uae : data) {
             if (uae.getIsDeleted() == 1) {
                 deleteId.add(uae.getUnitId());
             } else if (uae.getIsDeleted() == 0) {
                 updateList.add(uae);
             }
         }
-        deleteById("unitId", deleteId);
+        // 先删除
+        deleteByQuery(key, deleteId);
         if (updateList.isEmpty()) {
             return;
         }
-        for (UnitAgencyMergeEntity entity : updateList) {
+        // 更新数据
+        List<UnitAgencyMergeEntity> addData = updateDataByQuery(key, updateList);
 
-            Map<String, Object> map = objectMapper.convertValue(entity, new TypeReference<Map<String, Object>>() {
+        // 添加数据
+        addElasticData(addData);
+    }
+
+
+    @Override
+    public void deleteByQuery(String key, List<Long> ids) throws IOException {
+        if (ids == null || ids.isEmpty()) {
+            return;
+        }
+        for (Long id : ids) {
+            DeleteByQueryRequest request = new DeleteByQueryRequest(INDEX);
+            request.setQuery(QueryBuilders.termQuery(key, id));
+            restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
+        }
+    }
+
+    @Override
+    public List<UnitAgencyMergeEntity> updateDataByQuery(String key, List<UnitAgencyMergeEntity> data) throws IOException {
+        List<UnitAgencyMergeEntity> addList = new ArrayList<>();
+
+        for (UnitAgencyMergeEntity item : data) {
+
+            Map<String, Object> params = objectMapper.convertValue(item, new TypeReference<Map<String, Object>>() {
                 @Override
                 public Type getType() {
                     return super.getType();
                 }
             });
 
-            map.remove("unitId");
-
             UpdateByQueryRequest request = new UpdateByQueryRequest(INDEX);
-            request.setQuery(QueryBuilders.termQuery("unitId", entity.getUnitId()));
+            request.setQuery(QueryBuilders.termQuery(key, params.get(key)));
+
+            request.setScript(new Script(ScriptType.INLINE, "painless", getPainlessCode(params), params));
+
+            BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
+
+            logger.info("response:{}", response);
 
+            if (response.getUpdated() != 0L) {
+                addList.add(item);
+            }
         }
+
+        return addList;
     }
 
     @Override
-    public void updateAgency(List<UnitAgencyMergeEntity> list) throws IOException {
-        List<UnitAgencyMergeEntity> updateList = new ArrayList<>();
-        List<Long> deleteId = new ArrayList<>();
-        for (UnitAgencyMergeEntity uae : list) {
-            if (uae.getIsDeleted() == 1) {
-                deleteId.add(uae.getAgencyId());
-            } else if (uae.getIsDeleted() == 0) {
-                updateList.add(uae);
-            }
+    public void addElasticData(List<UnitAgencyMergeEntity> data) throws IOException {
+        if (data.isEmpty()) {
+            return;
         }
-        deleteById("agencyId", deleteId);
-        list.clear();
-        updateElasticData("agencyId", updateList);
-    }
+        final BeanCopier beanCopier = BeanCopier.create(UnitAgencyMergeEntity.class, UnitAgencyDetailEntity.class, false);
+
+        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);
 
-    private void updateElasticData(String key, List<UnitAgencyMergeEntity> list) throws IOException {
-        if (list == null || list.isEmpty()) {
+            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);
+
+            saveResult.add(map);
+
+        }
+        if (saveResult.isEmpty()) {
             return;
         }
-        for (UnitAgencyMergeEntity item : list) {
-            UpdateByQueryRequest request = new UpdateByQueryRequest(INDEX);
-            request.setQuery(QueryBuilders.termQuery(key, item.getAgencyId()));
+        try {
+            BulkRequest bulkRequest = new BulkRequest();
+
+            for (Map<String, Object> val : saveResult) {
+                IndexRequest indexRequest = createIndexRequest(INDEX, val);
+                bulkRequest.add(indexRequest);
+            }
+            BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
+            logger.info("bulk insert status {}", bulkResponse.status());
+        } catch (IOException e) {
+            e.printStackTrace();
         }
+
     }
 
-    private void deleteById(String key, List<Long> deleteId) throws IOException {
-        if (deleteId == null || deleteId.isEmpty()) {
-            return;
-        }
-        for (Long id : deleteId) {
-            DeleteByQueryRequest request = new DeleteByQueryRequest(INDEX);
-            request.setQuery(QueryBuilders.termQuery(key, id));
-            restHighLevelClient.deleteByQuery(request, RequestOptions.DEFAULT);
-        }
+
+    private IndexRequest createIndexRequest(String index, Map<String, Object> value) {
+        IndexRequest indexRequest = new IndexRequest(index);
+        indexRequest.id(SnowFlake.nextId() + "");
+        indexRequest.source(value, XContentType.JSON);
+        return indexRequest;
     }
 
+
+    private String getPainlessCode(Map<String, Object> params) {
+        String[] items = new String[params.size()];
+        int index = 0;
+        for (String key : params.keySet()) {
+            items[index++] = "ctx._source." + key + "=params." + key;
+        }
+        return String.join(";", items);
+    }
 }