0


springboot 2.0集成elasticsearch 7.6.2(集群)

小伙伴们,你们好呀,我是老寇,跟我一起学习es 7.6.2

注:请点击我,获取源码

一、引入依赖配置pom.xml

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
        </dependency> 
       <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.6.2</version>
            <exclusions>
                <exclusion>
                    <groupId>org.elasticsearch</groupId>
                    <artifactId>elasticsearch</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>elasticsearch-rest-client</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>7.6.2</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.6.2</version>
        </dependency>

二、配置application-dev.yml(生产就克隆application-dev改成生产配置)

elasticsearch:
  host: 192.168.1.1:9200,192.1.2.133:9200,192.168.1.3:9200
  cluster-name: laokou-elasticsearch
  username:
  password:
  synonym:
    path: http://192.168.1.1:9048/laokou-service/synonym

问题思考:比如说,一条文章记录,它有标题,内容,阅读量,在数据存入es时,我需要对es配置分词器,并且能够通过阅读量来筛选数据,你怎么做?

三、配置ES注解

注解可以修饰属性或方法(前提是先配置)

type > 需要在es配置什么类型

participle > 需要配置什么分词器

/**
 * @author Kou Shenhai
 */
@Target({ElementType.FIELD,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
public @interface FieldInfo {

    /**
     * 默认 keyword
     * @return
     */
    String type() default "keyword";

    /**
     * 0 not_analyzed 1 ik_smart 2.ik_max_word 3.ik-index(自定义分词器)
     * @return
     */
    int participle() default 0;

}

拼接属性对应的类型及分词器

/**
 * 属性、类型、分词器
 * @author Kou Shenhai [email protected]
 * @version 1.0
 * @date 2021/2/9 0009 上午 10:20
 */
@Data
@NoArgsConstructor
public class FieldMapping {

    private String field;

    private String type;

    private Integer participle;

    public FieldMapping(String field, String type, Integer participle) {
        this.field = field;
        this.type = type;
        this.participle = participle;
    }
}

组装每个属性对应的类型及分词器 => List<FieldMapping>

/**
 * 每个属性对应的类型及分词器
 * @author Kou Shenhai [email protected]
 * @version 1.0
 * @date 2021/1/24 0024 下午 7:51
 */
@Slf4j
public class FieldMappingUtil {

    public static List<FieldMapping> getFieldInfo(Class clazz) {
        return getFieldInfo(clazz, null);
    }

    public static List<FieldMapping> getFieldInfo(Class clazz, String fieldName) {
        //返回class中的所有字段(包括私有字段)
        Field[] fields = clazz.getDeclaredFields();
        //创建FieldMapping集合
        List<FieldMapping> fieldMappingList = new ArrayList<>();
        for (Field field : fields) {
            //获取字段上的FieldInfo对象
            boolean annotationPresent = field.isAnnotationPresent(FieldInfo.class);
            if (annotationPresent) {
                FieldInfo fieldInfo = field.getAnnotation(FieldInfo.class);
                //获取字段名称
                String name = field.getName();
                fieldMappingList.add(new FieldMapping(name, fieldInfo.type(), fieldInfo.participle()));
            } else {
                continue;
            }
        }
        return fieldMappingList;
    }
}

四、配置es及swagger

/**
 * es配置文件
 * @author Kou Shenhai [email protected]
 * @version 1.0
 * @date 2020/8/9 0009 下午 2:01
 */
@Configuration
public class ElasticsearchConfig {
    
    private static final String HTTP_SCHEME = "http";

    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfig.class);
    /**
     * 权限验证
     */
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    /**
     * es主机
     */
    @Value("${elasticsearch.host}")
    private String[] host;

    @Value("${elasticsearch.username}")
    private String username;

    @Value("${elasticsearch.password}")
    private String password;

    @Bean
    public RestClientBuilder restClientBuilder() {
        HttpHost[] hosts = Arrays.stream(host)
                .map(this::makeHttpHost)
                .filter(Objects::nonNull)
                .toArray(HttpHost[]::new);
        LOGGER.info("host:{}",Arrays.toString(hosts));
        //配置权限验证
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));
        return RestClient.builder(hosts).setHttpClientConfigCallback(httpClientBuilder ->
                httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
                .setMaxConnPerRoute(100)
                //最大连接数
                .setMaxConnTotal(100)
        ).setRequestConfigCallback(builder -> {
            builder.setConnectTimeout(-1);
            builder.setSocketTimeout(60000);
            builder.setConnectionRequestTimeout(-1);
            return builder;
        });
    }

    /**
     * 处理请求地址
     * @param address
     * @return
     */
    private HttpHost makeHttpHost(String address) {
        assert StringUtils.isNotEmpty(address);
        String[] hostAddress = address.split(":");
        if (hostAddress.length == 2) {
            String ip = hostAddress[0];
            Integer port = Integer.valueOf(hostAddress[1]);
            return new HttpHost(ip, port, HTTP_SCHEME);
        } else {
            return null;
        }
    }

    /**
     * 配置highLevelClient bean
     * @param restClientBuilder
     * @return
     */
    @Bean(name = "restHighLevelClient")
    public RestHighLevelClient restHighLevelClient(@Autowired RestClientBuilder restClientBuilder) {
        return new RestHighLevelClient(restClientBuilder);
    }
}
/**
 * @author Kou Shenhai
 */
@Configuration
@EnableSwagger2
public class SwaggerConfig {
    @Bean
    public Docket createRestApi() {
        return new Docket(DocumentationType.SWAGGER_2)
                .apiInfo(apiInfo())
                .select()
                .apis(RequestHandlerSelectors.withMethodAnnotation(ApiOperation.class))
                .paths(PathSelectors.any())
                .build();
    }
    private ApiInfo apiInfo() {
        return new ApiInfoBuilder()
                .title("API文档")
                .version("2.0.0")
                .description("API文档 - Elasticsearch服务")
                //作者信息
                .contact(new Contact("寇申海", "https://blog.csdn.net/qq_39893313", "[email protected]"))
                .build();
    }
}

五、ES工具类 (索引相关配置不懂的,请查看elasticsearch 7.6.2 - 索引管理)

/**
 * Elasticsearch工具类-用于操作ES
 * @author Kou Shenhai [email protected]
 * @version 1.0
 * @date 2021/1/24 0024 下午 5:42
 */
@Slf4j
@Component
public class ElasticsearchUtil {

    private static final String PRIMARY_KEY_NAME = "id";

    @Value("${elasticsearch.synonym.path}")
    private String synonymPath;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 批量同步数据到ES
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param jsonDataList 数据列表
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean saveDataBatch(String indexName,String indexAlias,String jsonDataList,Class clazz) throws IOException {
        //判空
        if (StringUtils.isBlank(jsonDataList)) {
            return false;
        }
        if (syncIndex(indexName, indexAlias, clazz)) {
            return false;
        }
        //3.批量操作Request
        BulkRequest bulkRequest = packBulkIndexRequest(indexName, jsonDataList);
        if (bulkRequest.requests().isEmpty()) {
            return false;
        }
        final BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (bulk.hasFailures()) {
            for (BulkItemResponse item : bulk.getItems()) {
                log.error("索引[{}],主键[{}]更新操作失败,状态为:[{}],错误信息:{}",indexName,item.getId(),item.status(),item.getFailureMessage());
            }
            return false;
        }
        // 记录索引新增与修改数量
        Integer createdCount = 0;
        Integer updatedCount = 0;
        for (BulkItemResponse item : bulk.getItems()) {
            if (DocWriteResponse.Result.CREATED.equals(item.getResponse().getResult())) {
                createdCount++;
            } else if (DocWriteResponse.Result.UPDATED.equals(item.getResponse().getResult())){
                updatedCount++;
            }
        }
        log.info("索引[{}]批量同步更新成功,共新增[{}]个,修改[{}]个",indexName,createdCount,updatedCount);
        return true;
    }

    /**
     * ES修改数据
     * @param indexName 索引名称
     * @param id 主键
     * @param paramJson 参数JSON
     * @return
     */
    public boolean updateData(String indexName,String id,String paramJson) {
        UpdateRequest updateRequest = new UpdateRequest(indexName, id);
        //如果修改索引中不存在则进行新增
        updateRequest.docAsUpsert(true);
        //立即刷新数据
        updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        updateRequest.doc(paramJson,XContentType.JSON);
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            log.info("索引[{}],主键:【{}】操作结果:[{}]",indexName,id,updateResponse.getResult());
            if (DocWriteResponse.Result.CREATED.equals(updateResponse.getResult())) {
                //新增
                log.info("索引:【{}】,主键:【{}】新增成功",indexName,id);
                return true;
            } else if (DocWriteResponse.Result.UPDATED.equals(updateResponse.getResult())) {
                //修改
                log.info("索引:【{}】,主键:【{}】修改成功",indexName, id);
                return true;
            } else if (DocWriteResponse.Result.NOOP.equals(updateResponse.getResult())) {
                //无变化
                log.info("索引:[{}],主键:[{}]无变化",indexName, id);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("索引:[{}],主键:【{}】,更新异常:[{}]",indexName, id,e);
            return false;
        }
        return false;
    }

    /**
     * 批量修改ES
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param jsonDataList 数据列表
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean updateDataBatch(String indexName,String indexAlias, String jsonDataList,Class clazz) throws IOException {
        //1.创建索引
        boolean createIndexFlag = createIndex(indexName,indexAlias, clazz);
        if (!createIndexFlag) {
            return false;
        }
        return this.updateDataBatch(indexName,jsonDataList);
    }

    /**
     * 删除数据
     * @param indexName 索引名称
     * @param id 主键
     * @return
     */
    public boolean deleteData(String indexName,String id) {
        DeleteRequest deleteRequest = new DeleteRequest(indexName);
        deleteRequest.id(id);
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            if (DocWriteResponse.Result.NOT_FOUND.equals(deleteResponse.getResult())) {
                log.error("索引:【{}】,主键:【{}】删除失败",indexName, id);
                return false;
            } else {
                log.info("索引【{}】主键【{}】删除成功",indexName, id);
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除索引【{}】出现异常[{}]",indexName,e);
            return false;
        }
    }

    /**
     * 批量删除ES
     * @param indexName 索引名称
     * @param ids id列表
     * @return
     */
    public boolean deleteDataBatch(String indexName,List<String> ids) {
        if (CollectionUtils.isEmpty(ids)) {
            return false;
        }
        BulkRequest bulkRequest = packBulkDeleteRequest(indexName, ids);
        try {
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                for (BulkItemResponse item : bulk.getItems()) {
                    log.error("删除索引:[{}],主键:{}失败,信息:{}",indexName,item.getId(),item.getFailureMessage());
                }
                return false;
            }
            //记录索引新增与修改数量
            Integer deleteCount = 0;
            for (BulkItemResponse item : bulk.getItems()) {
                if (DocWriteResponse.Result.DELETED.equals(item.getResponse().getResult())) {
                    deleteCount++;
                }
            }
            log.info("批量删除索引[{}]成功,共删除[{}]个",indexName,deleteCount);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除索引:【{}】出现异常:{}",indexName,e);
            return false;
        }
    }

    /**
     * 组装删除操作
     * @param indexName 索引名称
     * @param ids id列表
     * @return
     */
    private BulkRequest packBulkDeleteRequest(String indexName, List<String> ids) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        ids.forEach(id -> {
            DeleteRequest deleteRequest = new DeleteRequest(indexName);
            deleteRequest.id(id);
            bulkRequest.add(deleteRequest);
        });
        return bulkRequest;
    }

    /**
     * 批量修改ES
     * @param indexName 索引名称
     * @param jsonDataList json数据列表
     * @return
     */
    public boolean updateDataBatch(String indexName, String jsonDataList) {
        //判空
        if (StringUtils.isBlank(jsonDataList)) {
            return false;
        }
        BulkRequest bulkRequest = packBulkUpdateRequest(indexName, jsonDataList);
        if (bulkRequest.requests().isEmpty()) {
            return false;
        }
        try {
            //同步执行
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulk.hasFailures()) {
                for (BulkItemResponse item : bulk.getItems()) {
                    log.error("索引【{}】,主键[{}]修改操作失败,状态为【{}】,错误信息:{}",indexName,item.status(),item.getFailureMessage());
                }
                return false;
            }
            //记录索引新增与修改数量
            Integer createCount = 0;
            Integer updateCount = 0;
            for (BulkItemResponse item : bulk.getItems()) {
                if (DocWriteResponse.Result.CREATED.equals(item.getResponse().getResult())) {
                    createCount++;
                } else if (DocWriteResponse.Result.UPDATED.equals(item.getResponse().getResult())) {
                    updateCount++;
                }
            }
            log.info("索引【{}】批量修改更新成功,共新增[{}]个,修改[{}]个",indexName,createCount,updateCount);
        } catch (IOException e) {
            e.printStackTrace();
            log.error("索引【{}】批量修改更新出现异常",indexName);
            return false;
        }
        return true;
    }

    /**
     * 组装bulkUpdate
     * @param indexName 索引名称
     * @param jsonDataList 数据列表
     * @return
     */
    private BulkRequest packBulkUpdateRequest(String indexName,String jsonDataList) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        JSONArray jsonArray = JSONArray.parseArray(jsonDataList);
        if (jsonArray.isEmpty()) {
            return bulkRequest;
        }
        jsonArray.forEach(o -> {
            Map<String, String> map = (Map<String, String>) o;
            UpdateRequest updateRequest = new UpdateRequest(indexName,map.get(PRIMARY_KEY_NAME));
            // 修改索引中不存在就新增
            updateRequest.docAsUpsert(true);
            updateRequest.doc(JSON.toJSONString(o),XContentType.JSON);
            bulkRequest.add(updateRequest);
        });
        return bulkRequest;
    }

    /**
     * 删除索引、新建索引
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    private boolean syncIndex(String indexName, String indexAlias, Class clazz) throws IOException {
        //1.删除索引
        boolean deleteAllFlag = deleteIndex(indexName);
        if (!deleteAllFlag) {
            return true;
        }
        //2.创建索引
        boolean createIndexFlag = createIndex(indexName, indexAlias, clazz);
        if (!createIndexFlag) {
            return true;
        }
        return false;
    }

    /**
     * 根据主键查询ES
     * @param indexName 索引名称
     * @param id 主键
     * @return
     */
    public String getDataById(String indexName,String id) {
        //判断索引是否存在
        //1.判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            return "";
        }
        GetRequest getRequest = new GetRequest(indexName, id);
        try {
            GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            String resultJson = getResponse.getSourceAsString();
            log.info("索引【{}】主键【{}】,查询结果:【{}】",indexName,id,resultJson);
            return resultJson;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("索引【{}】主键[{}],查询异常:{}",indexName,id,e);
            return "";
        }

    }

    /**
     * 清空索引内容
     * @param indexName 索引名称
     * @return
     */
    public boolean deleteAll(String indexName) {
        //1.判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            log.error("索引【{}】不存在,删除失败",indexName);
            return false;
        }
        DeleteRequest deleteRequest = new DeleteRequest(indexName);
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        try {
            DeleteResponse deleteResponse = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            if (DocWriteResponse.Result.NOT_FOUND.equals(deleteResponse.getResult())) {
                log.error("索引【{}】删除失败",indexName);
                return false;
            }
            log.info("索引【{}】删除成功",indexName);
            return true;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("删除索引[{}],出现异常[{}]",indexName,e);
            return false;
        }
    }

    /**
     * 批量数据保存到ES-异步
     * @param indexName 索引名称
     * @param indexAlias 索引别名
     * @param jsonDataList 数据列表
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean saveDataBatchSync(String indexName,String indexAlias,String jsonDataList,Class clazz) throws IOException {
        //判空
        if (StringUtils.isBlank(jsonDataList)) {
            return false;
        }
        if (syncIndex(indexName, indexAlias, clazz)) {
            return false;
        }
        //3.批量操作Request
        BulkRequest bulkRequest = packBulkIndexRequest(indexName, jsonDataList);
        if (bulkRequest.requests().isEmpty()) {
            return false;
        }
        //4.异步执行
        ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                if (bulkItemResponses.hasFailures()) {
                    for (BulkItemResponse item : bulkItemResponses.getItems()) {
                        log.error("索引【{}】,主键【{}】更新失败,状态【{}】,错误信息:{}",indexName,item.getId(),
                                item.status(),item.getFailureMessage());
                    }
                }
            }
            //失败操作
            @Override
            public void onFailure(Exception e) {
                log.error("索引【{}】批量异步更新出现异常:{}",indexName,e);
            }
        };
        restHighLevelClient.bulkAsync(bulkRequest,RequestOptions.DEFAULT,listener);
        log.info("索引批量更新索引【{}】中",indexName);
        return true;
    }

    /**
     * 删除索引
     * @param indexName 索引名称
     * @return
     * @throws IOException
     */
    public boolean deleteIndex(String indexName) throws IOException {
        //1.判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            log.error("索引【{}】不存在,删除失败",indexName);
            return false;
        }
        //2.删除操作Request
        DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
        deleteIndexRequest.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
        AcknowledgedResponse acknowledgedResponse = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        if (!acknowledgedResponse.isAcknowledged()) {
            log.error("索引【{}】删除失败",indexName);
            return false;
        }
        log.info("索引【{}】删除成功",indexName);
        return true;
    }

    /**
     * 批量操作的Request
     * @param indexName 索引名称
     * @param jsonDataList json数据列表
     * @return
     */
    private BulkRequest packBulkIndexRequest(String indexName,String jsonDataList) {
        BulkRequest bulkRequest = new BulkRequest();
        //IMMEDIATE > 请求向es提交数据,立即进行数据刷新<实时性高,资源消耗高>
        //WAIT_UNTIL >  请求向es提交数据,等待数据完成刷新<实时性高,资源消耗低>
        //NONE > 默认策略<实时性低>
        bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        JSONArray jsonArray = JSONArray.parseArray(jsonDataList);
        if (jsonArray.isEmpty()) {
            return bulkRequest;
        }
        //循环数据封装bulkRequest
        jsonArray.forEach(obj ->{
            final Map<String, String> map = (Map<String, String>) obj;
            IndexRequest indexRequest = new IndexRequest(indexName);
            indexRequest.source(JSON.toJSONString(obj),XContentType.JSON);
            indexRequest.id(map.get(PRIMARY_KEY_NAME));
            bulkRequest.add(indexRequest);
        });
        return bulkRequest;
    }

    /**
     * 创建索引
     * @param indexName 索引名称
     * @param indexAlias 别名
     * @param clazz 类型
     * @return
     * @throws IOException
     */
    public boolean createIndex(String indexName,String indexAlias,Class clazz) throws IOException {
        //判断索引是否存在
        boolean result = isIndexExists(indexName);
        if (!result) {
            boolean createResult = createIndexAndCreateMapping(indexName,indexAlias, FieldMappingUtil.getFieldInfo(clazz));
            if (!createResult) {
                log.info("索引【{}】创建失败",indexName);
                return false;
            }
        }
        log.info("索引:[{}]创建成功",indexName);
        return true;
    }

    /**
     * 数据同步到ES
     * @param id 主键
     * @param indexName 索引名称
     * @param jsonData json数据
     * @param clazz 类型
     * @return
     */
    public boolean saveData(String id,String indexName,String indexAlias,String jsonData,Class clazz) throws IOException {
        //1.创建索引
        boolean createIndexFlag = createIndex(indexName,indexAlias, clazz);
        if (!createIndexFlag) {
            return false;
        }
        //2.创建操作Request
        IndexRequest indexRequest = new IndexRequest(indexName);
        //3.配置相关信息
        indexRequest.source(jsonData, XContentType.JSON);
        //IMMEDIATE > 立即刷新
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        indexRequest.id(id);
        IndexResponse response = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        //4.判断索引是新增还是修改
        if (IndexResponse.Result.CREATED.equals(response.getResult())) {
            log.info("索引【{}】保存成功",indexName);
            return true;
        } else if (IndexResponse.Result.UPDATED.equals(response.getResult())) {
            log.info("索引【{}】修改成功",indexName);
            return true;
        }
        return false;
    }

    /**
     * 判断索引是否存在
     * @param indexName 索引名称
     * @return
     */
    public boolean isIndexExists(String indexName) {
        try {
            GetIndexRequest getIndexRequest = new GetIndexRequest(indexName);
            return restHighLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
        }catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 创建索引设置相关配置信息
     * @param indexName 索引名称
     * @param indexAlias 索引别名
     * @param fieldMappingList 数据列表
     * @return
     * @throws IOException
     */
    private boolean createIndexAndCreateMapping(String indexName,String indexAlias, List<FieldMapping> fieldMappingList) throws IOException {
        //封装es索引的mapping
        XContentBuilder mapping = packEsMapping(fieldMappingList, null);
        mapping.endObject().endObject();
        mapping.close();
        //进行索引的创建
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
        //配置分词器
        XContentBuilder settings = packSettingMapping();
        XContentBuilder aliases = packEsAliases(indexAlias);
        log.info("索引配置脚本:{}",settings);
        log.info("索引字段内容:{}",mapping);
        createIndexRequest.settings(settings);
        createIndexRequest.mapping("_doc", mapping);
        createIndexRequest.aliases(aliases);
        //同步方式创建索引
        CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(createIndexRequest,RequestOptions.DEFAULT);
        boolean acknowledged = createIndexResponse.isAcknowledged();
        if (acknowledged) {
            log.info("索引:{}创建成功", indexName);
            return true;
        } else {
            log.error("索引:{}创建失败", indexName);
            return false;
        }
    }

    /**
     * 配置ES别名
     * @author Kou Shenhai
     * @param alias 别名
     * @return
     * @throws IOException
     */
    private XContentBuilder packEsAliases(String alias) throws IOException{
        XContentBuilder aliases = XContentFactory.jsonBuilder().startObject()
                .startObject(alias).endObject();
        aliases.endObject();
        aliases.close();
        return aliases;
    }

    /**
     * 配置Mapping
     * @param fieldMappingList 组装的实体类信息
     * @param mapping
     * @return
     * @throws IOException
     */
    private XContentBuilder packEsMapping(List<FieldMapping> fieldMappingList,XContentBuilder mapping) throws IOException {
        if (mapping == null) {
            //如果对象是空,首次进入,设置开始结点
            mapping = XContentFactory.jsonBuilder().startObject()
                    .field("dynamic",true)
                    .startObject("properties");
        }
        //循环实体对象的类型集合封装ES的Mapping
        for (FieldMapping fieldMapping : fieldMappingList) {
            String field = fieldMapping.getField();
            String dataType = fieldMapping.getType();
            Integer participle = fieldMapping.getParticiple();
            //设置分词规则
            if (Constant.NOT_ANALYZED.equals(participle)) {
                if (FieldTypeEnum.DATE.getValue().equals(dataType)) {
                    mapping.startObject(field)
                            .field("type", dataType)
                            .field("format","yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis")
                            .endObject();
                } else {
                    mapping.startObject(field)
                            .field("type", dataType)
                            .endObject();
                }
            } else if (Constant.IK_INDEX.equals(participle)) {
                mapping.startObject(field)
                        .field("type",dataType)
                        .field("eager_global_ordinals",true)
                        //fielddata=true 用来解决text字段不能进行聚合操作
                        .field("fielddata",true)
                        .field("boost",100.0)
                        .field("analyzer","ik-index-synonym")
                        .field("search_analyzer","ik-search-synonym")
                        .startObject("fields").startObject("data-pinyin")
                        .field("term_vector", "with_positions_offsets")
                        .field("analyzer","ik-search-pinyin")
                        .field("type",dataType)
                        .field("boost",100.0)
                        .endObject().endObject()
                        .endObject();
            }
        }
        return mapping;
    }

    /**
     * 配置Settings
     * @return
     * @throws IOException
     */
    private XContentBuilder packSettingMapping() throws IOException {
        XContentBuilder setting = XContentFactory.jsonBuilder().startObject()
                .startObject("index")
                .field("number_of_shards",5)
                .field("number_of_replicas",1)
                .field("refresh_interval","120s")
                .endObject()
                .startObject("analysis");
        //ik分词 同义词 拼音
        setting.startObject("analyzer")
                .startObject("ik-search-pinyin")
                .field("type","custom")
                .field("tokenizer","ik_smart")
                .field("char_filter",new String[] {"html_strip"})
                .field("filter", new String[]{"laokou-pinyin","word_delimiter","lowercase", "asciifolding"})
                .endObject();
        setting.startObject("ik-index-synonym")
                .field("type","custom")
                .field("tokenizer","ik_max_word")
                .field("char_filter",new String[] {"html_strip"})
                .field("filter", new String[]{"laokou-remote-synonym"})
                .endObject();
        setting.startObject("ik-search-synonym")
                .field("type","custom")
                .field("tokenizer","ik_smart")
                .field("char_filter",new String[] {"html_strip"})
                .field("filter", new String[]{"laokou-remote-synonym"})
                .endObject();
        setting.endObject();
        //设置拼音分词器 同义词分词
        setting.startObject("filter")
                .startObject("laokou-pinyin")
                .field("type", "pinyin")
                .field("keep_first_letter", false)
                .field("keep_separate_first_letter", false)
                .field("keep_full_pinyin", true)
                .field("keep_original", false)
                .field("keep_joined_full_pinyin",true)
                .field("limit_first_letter_length", 16)
                .field("lowercase", true)
                .field("remove_duplicated_term", true)
                .endObject()
                .startObject("laokou-remote-synonym")
                .field("type","dynamic_synonym")
                .field("synonyms_path", synonymPath)
                .field("interval",120)
                .field("dynamic_reload",true)
                .endObject()
                .endObject();
        setting.endObject().endObject();
        setting.close();
        return setting;
    }
}

问题思考:比如说,我有几条记录,文章记录,聊天记录,订单记录,它们是不同的索引,需要单独建立索引,怎么根据不同的数据类型来创建不同的索引?你会怎么做?

六、索引管理工具类

/**
 * 索引管理
 * @author Kou Shenhai [email protected]
 * @version 1.0
 * @date 2021/10/31 0031 上午 10:11
 */
public class FieldUtil {

    public static final String MESSAGE_INDEX = "message";

    private static final Map<String,Class<?>> classMap = new HashMap<>(16);

    static {
        classMap.put(FieldUtil.MESSAGE_INDEX, MessageIndex.class);
    }

    public static Class<?> getClazz(final String indexName) {
        return classMap.getOrDefault(indexName,Object.class);
    }
}

七、测试es

/**
 * Elasticsearch API 服务
 * @author Kou Shenhai [email protected]
 * @version 1.0
 * @date 2021/2/8 0008 下午 6:33
 */
@RestController
@RequestMapping("/api")
@Api(tags = "Elasticsearch API 服务")
public class ElasticsearchController {

    @Autowired
    private ElasticsearchUtil elasticsearchUtil;

    @PostMapping("/sync")
    @ApiOperation("同步数据到ES")
    @CrossOrigin
    public void syncIndex(@RequestBody final ElasticsearchModel model) throws IOException {
        String id = model.getId();
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonData = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.saveData(id,indexName,indexAlias,jsonData,clazz);
    }

    @PostMapping("/batchSync")
    @ApiOperation("批量数据保存到ES-异步")
    @CrossOrigin
    public void batchSyncIndex(@RequestBody final ElasticsearchModel model) throws IOException {
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonDataList = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.saveDataBatchSync(indexName,indexAlias,jsonDataList,clazz);
    }

    @PostMapping("/batch")
    @ApiOperation("批量同步数据到ES")
    @CrossOrigin
    public void saveBatchIndex(@RequestBody final ElasticsearchModel model) throws IOException {
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonDataList = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.saveDataBatch(indexName,indexAlias,jsonDataList,clazz);
    }

    @GetMapping("/get")
    @ApiOperation("根据主键获取ES")
    @CrossOrigin
    @ApiImplicitParams({
            @ApiImplicitParam(name = "indexName",value = "索引名称",required = true,paramType = "query",dataType = "String"),
            @ApiImplicitParam(name = "id",value = "主键",required = true,paramType = "query",dataType = "String")
    })
    public HttpResultUtil<String> getDataById(@RequestParam("indexName")String indexName,@RequestParam("id")String id) {
        return new HttpResultUtil<String>().ok(elasticsearchUtil.getDataById(indexName,id));
    }

    @PutMapping("/batch")
    @ApiOperation("批量修改ES")
    @CrossOrigin
    public void updateDataBatch(@RequestBody final ElasticsearchModel model) throws IOException {
        String indexName = model.getIndexName();
        String indexAlias = model.getIndexAlias();
        String jsonDataList = model.getData();
        Class<?> clazz = FieldUtil.getClazz(indexAlias);
        elasticsearchUtil.updateDataBatch(indexName,indexAlias,jsonDataList,clazz);
    }

    @PutMapping("/sync")
    @ApiOperation("同步修改ES")
    @CrossOrigin
    public void updateData(@RequestBody final ElasticsearchModel model) {
        String id = model.getId();
        String indexName = model.getIndexName();
        String paramJson = model.getData();
        elasticsearchUtil.updateData(indexName,id,paramJson);
    }

    @DeleteMapping("/batch")
    @ApiOperation("批量删除ES")
    @CrossOrigin
    public void deleteDataBatch(@RequestParam("indexName")String indexName,@RequestParam("ids")List<String> ids) {
        elasticsearchUtil.deleteDataBatch(indexName,ids);
    }

    @DeleteMapping("/sync")
    @ApiOperation("同步删除ES")
    @CrossOrigin
    public void deleteData(@RequestParam("indexName")String indexName,@RequestParam("id")String id) {
        elasticsearchUtil.deleteData(indexName,id);
    }

    @DeleteMapping("/all")
    @ApiOperation("清空ES")
    @CrossOrigin
    public void deleteAll(@RequestParam("indexName")String indexName) {
        elasticsearchUtil.deleteAll(indexName);
    }
}

大功告成

补充:可根据自己的业务进行数据分区


本文转载自: https://blog.csdn.net/qq_39893313/article/details/123013305
版权归原作者 k↑ 所有, 如有侵权,请联系我们删除。

“springboot 2.0集成elasticsearch 7.6.2(集群)”的评论:

还没有评论