基于ElasticsearchRepository进行简单封装实现非空更新，saveOrUpdate[笔记]

import xx.xx.xx.PageQuery;
import xx.xx.xx.SortParam;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.sort.SortBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.annotation.Id;
import org.springframework.data.domain.*;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.document;
import org.springframework.data.elasticsearch.core.query.BulkOptions;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.UpdateQuery;

import java.lang.reflect.Field;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public abstract class EsbaseServiceImpl>  implements IEsbaseService {

    @Autowired(required = false)
    public M repository;
    @Autowired
    public ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Override
    public  S save(S entity) {
        return repository.save(entity);
    }

    @Override
    public  Iterable saveAll(Iterable entities) {
        return repository.saveAll(entities);
    }

    @Override
    public Optional findById(ID id) {
        return repository.findById(id);
    }

    @Override
    public boolean existsById(ID id) {
        return repository.existsById(id);
    }

    @Override
    public Collection findAll() {
        return list(new PageQuery());
    }

    @Override
    public Collection findAllById(Collection ids) {
        return (Collection) repository.findAllById(ids);
    }

    @Override
    public long count() {
        return repository.count();
    }

    @Override
    public void deleteById(ID id) {
        repository.deleteById(id);
    }

    @Override
    public void delete(T entity) {
        final List fields = getEntityAllField();

        AtomicInteger num = new AtomicInteger();
        // 构建过滤条件
        BoolQueryBuilder filter = buildFilterBoolQueryBuilder(fields, entity, num);

        if (num.intValue() < 1) {
            return;
        }

        // 构建查询条件
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        queryBuilder.withFilter(filter);
        // 执行删除
        elasticsearchRestTemplate.delete(queryBuilder.build(), getEntityClass());
    }

    @Override
    public void deleteAllById(Iterable ids) {
        repository.deleteAllById(ids);
    }

    @Override
    public void deleteAll(Collection entities) {
        if (CollectionUtils.isEmpty(entities)) {
            return;
        }

        entities.forEach(this::delete);
    }

    @Override
    public void deleteAll() {
        repository.deleteAll();
    }

    @Override
    public Iterable findAll(Sort sort) {
        return repository.findAll(sort);
    }

    @Override
    public Page findAll(Pageable pageable) {
        return repository.findAll(pageable);
    }

    @Override
    public Page searchSimilar(T entity, String[] fields, Pageable pageable) {
        return repository.searchSimilar(entity, fields, pageable);
    }

    @Override
    public  Page search(Q query) {
        Long total = count(query);
        SearchHits searchHits = commonSearch(query, true);
        if (searchHits.getTotalHits() > 0) {
            List searchProductList = searchHits.stream().map(SearchHit::getContent).collect(Collectors.toList());
            return new PageImpl<>(searchProductList, PageRequest.of(query.getPage(), query.getSize()), total);
        }
        return new PageImpl(new ArrayList<>(), PageRequest.of(query.getPage(), query.getSize()), total);
    }

    @Override
    public  Long count(Q query) {
        return commonSearch(query, false).getTotalHits();
    }

    @Override
    public  List list(Q query) {
        SearchHits searchHits = commonSearch(query, false);
        if (searchHits.getTotalHits() > 0) {
            return searchHits.stream().map(SearchHit::getContent).collect(Collectors.toList());
        }
        return null;
    }

    @Override
    public  List list(Q query, String... columnName) {
        SearchHits searchHits = commonSearch(query, false, columnName);
        if (searchHits.getTotalHits() > 0) {
            return searchHits.stream().map(SearchHit::getContent).collect(Collectors.toList());
        }

        return null;
    }

    @Override
    public  void update(S entity) {
        commonUpdate(CollectionUtils.singleList(entity), false);
    }

    @Override
    public  void updateAndFlush(S entity) {
        commonUpdate(CollectionUtils.singleList(entity), true);
    }

    @Override
    public  void update(Collection entities) {
        commonUpdate(entities, false);
    }

    @Override
    public  void updateAndFlush(Collection entities) {
        commonUpdate(entities, true);
    }


    @Override
    public  void saveOrUpdate(Collection entities) {
        Map> tempMap = idTempMap(entities);

        if (tempMap == null) {
            return;
        }

        List ids = new ArrayList<>(tempMap.keySet());

        Collection records = findAllById(ids);

        if (CollectionUtils.isEmpty(records)) {
            saveAll(entities);
            return;
        }

        List save = new CopyOnWriteArrayList<>();

        List update = new CopyOnWriteArrayList<>();

        records.forEach(entity -> {
            Field[] declaredFields = entity.getClass().getDeclaredFields();
            for (Field field : declaredFields) {
                if (!field.isAnnotationPresent(Id.class)) {
                    continue;
                }

                ID id = (ID) doGetFieldValue(field, entity);

                Map map = tempMap.get(id);

                if (map == null) {
                    // save
                    save.add(entity);
                } else {
                    // update
                    update.add(entity);
                }
            }
        });

        if (CollectionUtils.notEmpty(save)) {
            saveAll(save);
        }

        if (CollectionUtils.notEmpty(update)) {
            commonUpdate(update, true);
        }
    }

    
    private  void commonUpdate(Collection entities, Boolean flush) {
        Map> tempMap = idTempMap(entities);

        if (tempMap == null) {
            return;
        }

        List queries = new CopyOnWriteArrayList<>();

        tempMap.forEach((id, params) -> {
            UpdateQuery build = UpdateQuery.builder(String.valueOf(id))
                    .withdocument(document.from(params))
                    .build();
            queries.add(build);
        });

        if (flush) {
            // 立刻刷新，损害性能
            elasticsearchRestTemplate.bulkUpdate(
                    queries,
                    BulkOptions.builder().withRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).build(),
                    elasticsearchRestTemplate.getIndexCoordinatesFor(getEntityClass()));

        } else {
            // 不执行立刻刷新，损害性能
            elasticsearchRestTemplate.bulkUpdate(queries, getEntityClass());
        }
    }

    private  Map> idTempMap(Collection entities) {
        if (CollectionUtils.isEmpty(entities)) {
            return null;
        }

        final List fields = getEntityAllField();

        Map> tempMap = new ConcurrentHashMap<>();

        entities.forEach(entity -> buildIdMapParams(fields, entity, tempMap));

        if (CollectionUtils.isEmpty(tempMap)) {
            return null;
        }

        return tempMap;
    }

    private  void buildIdMapParams(List fields, S entity, Map> tempMap) {
        // 用来存放参数
        Map params = new linkedHashMap<>();

        for (Field field : fields) {
            Object o = doGetFieldValue(field, entity);

            if (o == null) {
                continue;
            }

            params.put(field.getName(), o);

            if (field.isAnnotationPresent(Id.class)) {
                // 主键ID
                tempMap.put((ID) o, params);
            }
        }
    }


    private BoolQueryBuilder buildFilterBoolQueryBuilder(List fields, T entity, AtomicInteger num) {
        // 查询构建器
        BoolQueryBuilder filter = QueryBuilders.boolQuery();

        for (Field field : fields) {
            Object obj = doGetFieldValue(field, entity);

            if (obj == null) {
                continue;
            }

            // 计数器统计数量+1
            num.incrementAndGet();

            filter.must(QueryBuilders.termQuery(field.getName(), obj));
        }

        return filter;
    }

    
    private Object doGetFieldValue(Field field, T entity) {
        field.setAccessible(true);
        // 一般属性
        Object o = null;
        try {
            o = field.get(entity);
        } catch (IllegalAccessException e) {
            log.error("获取属性异常", e);
        }

        return o;
    }


    
    private  SearchHits commonSearch(Q query, boolean page, String... columnName) {
        // 构建查询条件
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder();
        // 查询构建器
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        // 构建过滤条件
        buildFilterCondition(builder, query);

        queryBuilder.withQuery(builder);

        List sorts = sortFields(query);

        if (CollectionUtils.notEmpty(sorts)) {
            for (SortParam sort : sorts) {
                queryBuilder.withSort(SortBuilders.fieldSort(sort.getFieldName()).order(sort.getOrder()));
            }
        }
        // 分页条件
        if (page) {
            queryBuilder.withPageable(PageRequest.of(query.getPage(), query.getSize()));
        }

        NativeSearchQuery nativeSearchQuery = queryBuilder.build();

        // 页面返回字段设置
        if (columnName != null && columnName.length > 0) {
            nativeSearchQuery.addFields(columnName);
        } else {
            nativeSearchQuery.addFields(returnFields());
        }
        
        // 使用ElasticsearchRestTemplate进行复杂查询
        return elasticsearchRestTemplate.search(nativeSearchQuery, this.getEntityClass());
    }
}

  
     public  void buildFilterCondition(BoolQueryBuilder filter, Q queryParam) {
     	// 强转为实际请求对象
     	XxxQuery query = (XxxQuery)queryParam;
     	// 根据实际参数构造查询条件
        // eg:
        // 带分词匹配
        // filter.must(QueryBuilders.matchQuery("xxx", query.getXxx()));
        // 不分词匹配
        // filter.must(QueryBuilders.termQuery("xxx", query.getXxx()));
        // 范围匹配
        // filter.must(QueryBuilders.rangeQuery("createTime").gte(query.getCreateTime() + " 00:00:00"));
    }